merge: #9954

9954: Fix backwards-incompatible protocol version bump r=oleschoenburg a=oleschoenburg

## Description

This splits the protocol based on usage and reverts the version bump in `protocol.xml`.

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #9949 



Co-authored-by: Ole Schönburg <ole.schoenburg@gmail.com>

broker/pom.xml

@@ -387,7 +387,7 @@
           <includePluginDependencies>true</includePluginDependencies>
           <mainClass>uk.co.real_logic.sbe.SbeTool</mainClass>
           <arguments>
-            <argument>${project.build.resources[0].directory}/management-schema.xml</argument>
+            <argument>${project.build.resources[0].directory}/broker-protocol.xml</argument>
           </arguments>
           <workingDirectory>${project.build.directory}/generated-sources</workingDirectory>
         </configuration>

broker/src/main/java/io/camunda/zeebe/broker/exporter/stream/ExporterPositionsMessage.java

@@ -7,9 +7,9 @@
  */
 package io.camunda.zeebe.broker.exporter.stream;
 
-import io.camunda.zeebe.clustering.management.ExporterPositionsDecoder;
-import io.camunda.zeebe.clustering.management.ExporterPositionsEncoder;
-import io.camunda.zeebe.clustering.management.ExporterPositionsEncoder.PositionsEncoder;
+import io.camunda.zeebe.broker.protocol.ExporterPositionsDecoder;
+import io.camunda.zeebe.broker.protocol.ExporterPositionsEncoder;
+import io.camunda.zeebe.broker.protocol.ExporterPositionsEncoder.PositionsEncoder;
 import io.camunda.zeebe.protocol.impl.encoding.SbeBufferWriterReader;
 import io.camunda.zeebe.util.buffer.BufferUtil;
 import java.util.HashMap;

broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/AdminApiRequestHandler.java

@@ -12,7 +12,7 @@
 import io.camunda.zeebe.broker.partitioning.PartitionManagerImpl;
 import io.camunda.zeebe.broker.transport.ApiRequestHandler;
 import io.camunda.zeebe.broker.transport.ErrorResponseWriter;
-import io.camunda.zeebe.protocol.record.AdminRequestType;
+import io.camunda.zeebe.protocol.management.AdminRequestType;
 import io.camunda.zeebe.transport.RequestType;
 import io.camunda.zeebe.transport.impl.AtomixServerTransport;
 import io.camunda.zeebe.util.Either;

broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/ApiRequestReader.java

@@ -8,8 +8,8 @@
 package io.camunda.zeebe.broker.transport.adminapi;
 
 import io.camunda.zeebe.broker.transport.ApiRequestHandler.RequestReader;
-import io.camunda.zeebe.protocol.record.AdminRequestDecoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderDecoder;
+import io.camunda.zeebe.protocol.management.AdminRequestDecoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderDecoder;
 import org.agrona.DirectBuffer;
 
 public class ApiRequestReader implements RequestReader<AdminRequestDecoder> {

broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/ApiResponseWriter.java

@@ -8,7 +8,7 @@
 package io.camunda.zeebe.broker.transport.adminapi;
 
 import io.camunda.zeebe.broker.transport.ApiRequestHandler.ResponseWriter;
-import io.camunda.zeebe.protocol.record.AdminResponseEncoder;
+import io.camunda.zeebe.protocol.management.AdminResponseEncoder;
 import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;
 import io.camunda.zeebe.transport.ServerOutput;
 import io.camunda.zeebe.transport.impl.ServerResponseImpl;

broker/src/main/java/io/camunda/zeebe/broker/transport/backupapi/BackupApiRequestHandler.java

@@ -13,8 +13,8 @@
 import io.camunda.zeebe.logstreams.log.LogStreamRecordWriter;
 import io.camunda.zeebe.protocol.impl.record.RecordMetadata;
 import io.camunda.zeebe.protocol.impl.record.value.management.CheckpointRecord;
-import io.camunda.zeebe.protocol.record.AdminRequestType;
-import io.camunda.zeebe.protocol.record.BackupRequestType;
+import io.camunda.zeebe.protocol.management.AdminRequestType;
+import io.camunda.zeebe.protocol.management.BackupRequestType;
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.intent.management.CheckpointIntent;

broker/src/main/java/io/camunda/zeebe/broker/transport/backupapi/BackupApiRequestReader.java

@@ -8,9 +8,9 @@
 package io.camunda.zeebe.broker.transport.backupapi;
 
 import io.camunda.zeebe.broker.transport.ApiRequestHandler.RequestReader;
-import io.camunda.zeebe.protocol.record.BackupRequestDecoder;
-import io.camunda.zeebe.protocol.record.BackupRequestType;
-import io.camunda.zeebe.protocol.record.MessageHeaderDecoder;
+import io.camunda.zeebe.protocol.management.BackupRequestDecoder;
+import io.camunda.zeebe.protocol.management.BackupRequestType;
+import io.camunda.zeebe.protocol.management.MessageHeaderDecoder;
 import org.agrona.DirectBuffer;
 
 public final class BackupApiRequestReader implements RequestReader<BackupRequestDecoder> {

broker/src/main/java/io/camunda/zeebe/broker/transport/backupapi/BackupApiResponseWriter.java

@@ -8,8 +8,8 @@
 package io.camunda.zeebe.broker.transport.backupapi;
 
 import io.camunda.zeebe.broker.transport.ApiRequestHandler.ResponseWriter;
-import io.camunda.zeebe.protocol.record.BackupResponseEncoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;
+import io.camunda.zeebe.protocol.management.BackupResponseEncoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderEncoder;
 import io.camunda.zeebe.transport.ServerOutput;
 import io.camunda.zeebe.transport.impl.ServerResponseImpl;
 import org.agrona.MutableDirectBuffer;

broker/src/main/java/io/camunda/zeebe/broker/transport/partitionapi/InterPartitionCommandReceiverImpl.java

@@ -10,8 +10,8 @@
 import io.atomix.cluster.MemberId;
 import io.camunda.zeebe.backup.processing.state.CheckpointState;
 import io.camunda.zeebe.broker.Loggers;
-import io.camunda.zeebe.clustering.management.InterPartitionMessageDecoder;
-import io.camunda.zeebe.clustering.management.MessageHeaderDecoder;
+import io.camunda.zeebe.broker.protocol.InterPartitionMessageDecoder;
+import io.camunda.zeebe.broker.protocol.MessageHeaderDecoder;
 import io.camunda.zeebe.logstreams.log.LogStreamRecordWriter;
 import io.camunda.zeebe.protocol.impl.record.RecordMetadata;
 import io.camunda.zeebe.protocol.impl.record.value.management.CheckpointRecord;

broker/src/main/java/io/camunda/zeebe/broker/transport/partitionapi/InterPartitionCommandSenderImpl.java

@@ -13,8 +13,8 @@
 import io.camunda.zeebe.backup.processing.state.CheckpointState;
 import io.camunda.zeebe.broker.Loggers;
 import io.camunda.zeebe.broker.partitioning.topology.TopologyPartitionListenerImpl;
-import io.camunda.zeebe.clustering.management.InterPartitionMessageEncoder;
-import io.camunda.zeebe.clustering.management.MessageHeaderEncoder;
+import io.camunda.zeebe.broker.protocol.InterPartitionMessageEncoder;
+import io.camunda.zeebe.broker.protocol.MessageHeaderEncoder;
 import io.camunda.zeebe.engine.transport.InterPartitionCommandSender;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.intent.Intent;

broker/src/main/resources/management-schema.xml → broker/src/main/resources/broker-protocol.xml

@@ -1,27 +1,11 @@
 <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
 <sbe:messageSchema xmlns:sbe="http://fixprotocol.io/2016/sbe"
   xmlns:xi="http://www.w3.org/2001/XInclude"
-  package="io.camunda.zeebe.clustering.management" id="5" version="2"
-  semanticVersion="0.1.0" description="Zeebe Cluster Management Protocol" byteOrder="littleEndian">
+  package="io.camunda.zeebe.broker.protocol" id="5" version="2"
+  semanticVersion="0.1.0" description="Zeebe Broker Communication Protocol" byteOrder="littleEndian" >
 
   <xi:include href="../../../../protocol/src/main/resources/common-types.xml"/>
 
-  <types>
-
-    <!-- binary data -->
-    <composite name="blob">
-      <type name="length" primitiveType="uint32" maxValue="2147483647"/>
-      <type name="varData" primitiveType="uint8" length="0"/>
-    </composite>
-
-    <enum name="errorResponseCode" encodingType="uint8" semanticType="String">
-      <validValue name="PARTITION_NOT_FOUND">0</validValue>
-      <validValue name="INVALID_PARAMETERS">1</validValue>
-      <validValue name="READ_ERROR">2</validValue>
-    </enum>
-
-  </types>
-
   <sbe:message id="1" name="ExporterPositions">
     <group id="0" name="positions">
       <field id="0" name="position" type="uint64" semanticType="long"/>
@@ -36,7 +20,7 @@
     <field name="recordKey" id="3" type="uint64" presence="optional"/>
     <field name="checkpointId" id="4" type="int64"/>
 
-    <data name="command" id="32" type="blob"/>
+    <data name="command" id="32" type="varDataEncoding"/>
   </sbe:message>
 
 </sbe:messageSchema>

broker/src/test/java/io/camunda/zeebe/broker/transport/adminapi/AdminApiRequestHandlerTest.java

@@ -18,7 +18,7 @@
 import io.camunda.zeebe.protocol.impl.encoding.AdminRequest;
 import io.camunda.zeebe.protocol.impl.encoding.AdminResponse;
 import io.camunda.zeebe.protocol.impl.encoding.ErrorResponse;
-import io.camunda.zeebe.protocol.record.AdminRequestType;
+import io.camunda.zeebe.protocol.management.AdminRequestType;
 import io.camunda.zeebe.protocol.record.ErrorCode;
 import io.camunda.zeebe.scheduler.testing.ControlledActorSchedulerRule;
 import io.camunda.zeebe.transport.ServerOutput;

broker/src/test/java/io/camunda/zeebe/broker/transport/backupapi/BackupApiRequestHandlerTest.java

@@ -16,7 +16,7 @@
 import io.camunda.zeebe.protocol.impl.encoding.AdminResponse;
 import io.camunda.zeebe.protocol.impl.encoding.BackupRequest;
 import io.camunda.zeebe.protocol.impl.encoding.ErrorResponse;
-import io.camunda.zeebe.protocol.record.BackupRequestType;
+import io.camunda.zeebe.protocol.management.BackupRequestType;
 import io.camunda.zeebe.protocol.record.ErrorCode;
 import io.camunda.zeebe.scheduler.testing.ControlledActorSchedulerExtension;
 import io.camunda.zeebe.transport.ServerOutput;

gateway/src/main/java/io/camunda/zeebe/gateway/admin/BrokerAdminRequest.java

@@ -11,9 +11,9 @@
 import io.camunda.zeebe.gateway.impl.broker.response.BrokerResponse;
 import io.camunda.zeebe.protocol.impl.encoding.AdminRequest;
 import io.camunda.zeebe.protocol.impl.encoding.AdminResponse;
-import io.camunda.zeebe.protocol.record.AdminRequestEncoder;
-import io.camunda.zeebe.protocol.record.AdminRequestType;
-import io.camunda.zeebe.protocol.record.AdminResponseEncoder;
+import io.camunda.zeebe.protocol.management.AdminRequestEncoder;
+import io.camunda.zeebe.protocol.management.AdminRequestType;
+import io.camunda.zeebe.protocol.management.AdminResponseEncoder;
 import io.camunda.zeebe.transport.RequestType;
 import io.camunda.zeebe.util.buffer.BufferWriter;
 import java.util.Optional;

protocol-impl/src/main/java/io/camunda/zeebe/protocol/impl/encoding/AdminRequest.java

@@ -7,11 +7,11 @@
  */
 package io.camunda.zeebe.protocol.impl.encoding;
 
-import io.camunda.zeebe.protocol.record.AdminRequestDecoder;
-import io.camunda.zeebe.protocol.record.AdminRequestEncoder;
-import io.camunda.zeebe.protocol.record.AdminRequestType;
-import io.camunda.zeebe.protocol.record.MessageHeaderDecoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;
+import io.camunda.zeebe.protocol.management.AdminRequestDecoder;
+import io.camunda.zeebe.protocol.management.AdminRequestEncoder;
+import io.camunda.zeebe.protocol.management.AdminRequestType;
+import io.camunda.zeebe.protocol.management.MessageHeaderDecoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderEncoder;
 import io.camunda.zeebe.util.buffer.BufferReader;
 import io.camunda.zeebe.util.buffer.BufferWriter;
 import org.agrona.DirectBuffer;

protocol-impl/src/main/java/io/camunda/zeebe/protocol/impl/encoding/AdminResponse.java

@@ -7,10 +7,10 @@
  */
 package io.camunda.zeebe.protocol.impl.encoding;
 
-import io.camunda.zeebe.protocol.record.AdminResponseDecoder;
-import io.camunda.zeebe.protocol.record.AdminResponseEncoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderDecoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;
+import io.camunda.zeebe.protocol.management.AdminResponseDecoder;
+import io.camunda.zeebe.protocol.management.AdminResponseEncoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderDecoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderEncoder;
 import io.camunda.zeebe.util.buffer.BufferReader;
 import io.camunda.zeebe.util.buffer.BufferWriter;
 import org.agrona.DirectBuffer;

protocol-impl/src/main/java/io/camunda/zeebe/protocol/impl/encoding/BackupRequest.java

@@ -7,11 +7,11 @@
  */
 package io.camunda.zeebe.protocol.impl.encoding;
 
-import io.camunda.zeebe.protocol.record.BackupRequestDecoder;
-import io.camunda.zeebe.protocol.record.BackupRequestEncoder;
-import io.camunda.zeebe.protocol.record.BackupRequestType;
-import io.camunda.zeebe.protocol.record.MessageHeaderDecoder;
-import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;
+import io.camunda.zeebe.protocol.management.BackupRequestDecoder;
+import io.camunda.zeebe.protocol.management.BackupRequestEncoder;
+import io.camunda.zeebe.protocol.management.BackupRequestType;
+import io.camunda.zeebe.protocol.management.MessageHeaderDecoder;
+import io.camunda.zeebe.protocol.management.MessageHeaderEncoder;
 import io.camunda.zeebe.util.buffer.BufferReader;
 import io.camunda.zeebe.util.buffer.BufferWriter;
 import org.agrona.DirectBuffer;

protocol/ignored-changes.json

@@ -38,6 +38,18 @@
           "justification": "Generated types are now annotated with a marker which is used for reflection; this only adds information but does not change semantics",
           "code": "java.annotation.added",
           "annotationType": "io.camunda.zeebe.protocol.record.ImmutableProtocol.Type"
+        },
+        {
+          "justification": "Ignore the removal of this old BpmnElementType that was used for testing.",
+          "code": "java.field.removed",
+          "classQualifiedName": "io.camunda.zeebe.protocol.record.value.BpmnElementType",
+          "fieldName": "TESTING_ONLY"
+        },
+        {
+          "justification": "Allow removal of AdminRequest and AdminResponse types",
+          "code": "java.class.removed",
+          "regex": true,
+          "classQualifiedName": "io.camunda.zeebe.protocol.record.Admin(Request|Response)\\w*"
         }
       ]
     }

protocol/pom.xml

@@ -25,7 +25,7 @@
   <properties>
     <version.java>8</version.java>
     <license.header>com/mycila/maven/plugin/license/templates/APACHE-2.txt</license.header>
-    <protocol.version>4</protocol.version>
+    <protocol.version>3</protocol.version>
   </properties>
 
   <dependencies>
@@ -179,6 +179,7 @@
           <mainClass>uk.co.real_logic.sbe.SbeTool</mainClass>
           <arguments>
             <argument>${project.build.outputDirectory}/protocol.xml</argument>
+            <argument>${project.build.outputDirectory}/cluster-management-protocol.xml</argument>
           </arguments>
           <workingDirectory>${project.build.directory}/generated-sources</workingDirectory>
         </configuration>

protocol/revapi.json

@@ -14,17 +14,6 @@
     "id": "differences",
     "configuration": {
       "differences": [
-        {
-          "justification": "Upgrading the protocol version should be backwards compatible",
-          "code": "java.field.constantValueChanged",
-          "classQualifiedName": "io.camunda.zeebe.protocol.Protocol",
-          "fieldName": "PROTOCOL_VERSION"
-        },
-        {
-          "justification": "Upgrading the schema version should be backwards compatible",
-          "code": "java.field.constantValueChanged",
-          "fieldName": "SCHEMA_VERSION"
-        },
         {
           "justification": "Ignore Enum order for BpmnElementType as ordinal() is not used and the elements are grouped in the enum.",
           "code": "java.field.enumConstantOrderChanged",
@@ -42,26 +31,6 @@
           "justification": "Ignore Enum order for ValueType as ordinal() is not used",
           "code": "java.field.enumConstantOrderChanged",
           "classQualifiedName": "io.camunda.zeebe.protocol.record.ValueType"
-        },
-        {
-          "justification": "Ignore the removal of this old BpmnElementType that was used for testing.",
-          "code": "java.field.removed",
-          "classQualifiedName": "io.camunda.zeebe.protocol.record.value.BpmnElementType",
-          "fieldName": "TESTING_ONLY"
-        },
-        {
-          "justification": "Ignore Enum order for AdminRequestType as ordinal() is not used",
-          "code": "java.field.enumConstantOrderChanged",
-          "classQualifiedName": "io.camunda.zeebe.protocol.record.AdminRequestType"
-        },
-        {
-          "justification": "Adding optional fields to the end of the root block is backwards compatible",
-          "code": "java.field.constantValueChanged",
-          "regex": true,
-          "classQualifiedName": "io.camunda.zeebe.protocol.record.AdminRequest(Decoder|Encoder)",
-          "fieldName": "BLOCK_LENGTH",
-          "oldValue": 3,
-          "newValue": 5
         }
       ]
     }

protocol/src/main/resources/cluster-management-protocol.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<sbe:messageSchema xmlns:sbe="http://fixprotocol.io/2016/sbe"
+  xmlns:xi="http://www.w3.org/2001/XInclude" package="io.camunda.zeebe.protocol.management"
+  id="1" version="0" semanticVersion="${project.version}"
+  description="Zeebe Cluster Management Protocol" byteOrder="littleEndian">
+
+  <xi:include href="common-types.xml"/>
+
+  <types>
+    <enum name="AdminRequestType" encodingType="uint8">
+      <validValue name="STEP_DOWN_IF_NOT_PRIMARY">0</validValue>
+      <validValue name="PAUSE_EXPORTING">1</validValue>
+    </enum>
+
+    <enum name="BackupRequestType" encodingType="uint8">
+      <validValue name="TAKE_BACKUP">0</validValue>
+    </enum>
+  </types>
+
+  <sbe:message name="AdminRequest" id="1">
+    <field name="partitionId" id="1" type="uint16"/>
+    <field name="type" id="2" type="AdminRequestType"/>
+    <field name="brokerId" id="3" type="uint16" presence="optional"/>
+  </sbe:message>
+
+  <sbe:message name="AdminResponse" id="2">
+  </sbe:message>
+
+  <sbe:message name="BackupRequest" id="3">
+    <field name="partitionId" id="1" type="uint16"/>
+    <field name="type" id="2" type="BackupRequestType"/>
+    <field name="backupId" id = "3" type="int64"/>
+  </sbe:message>
+
+  <sbe:message name="BackupResponse" id="4">
+  </sbe:message>
+</sbe:messageSchema>

protocol/src/main/resources/protocol.xml

@@ -80,14 +80,6 @@
       <type name="patchVersion" primitiveType="int32"/>
     </composite>
 
-    <enum name="AdminRequestType" encodingType="uint8">
-      <validValue name="STEP_DOWN_IF_NOT_PRIMARY">0</validValue>
-      <validValue name="PAUSE_EXPORTING">1</validValue>
-    </enum>
-
-    <enum name="BackupRequestType" encodingType="uint8">
-      <validValue name="TAKE_BACKUP">0</validValue>
-    </enum>
 
   </types>
 
@@ -129,23 +121,6 @@
     <data name="bpmnProcessId" id="1" type="varDataEncoding"/>
   </sbe:message>
 
-  <sbe:message name="AdminRequest" id="40">
-    <field name="partitionId" id="1" type="uint16"/>
-    <field name="type" id="2" type="AdminRequestType"/>
-    <field name="brokerId" id="3" type="uint16" presence="optional" sinceVersion="4"/>
-  </sbe:message>
-
-  <sbe:message name="AdminResponse" id="41">
-  </sbe:message>
-
-  <sbe:message name="BackupRequest" id="42">
-    <field name="partitionId" id="1" type="uint16"/>
-    <field name="type" id="2" type="BackupRequestType"/>
-    <field name="backupId" id = "3" type="int64"/>
-  </sbe:message>
-
-  <sbe:message name="BackupResponse" id="43">
-  </sbe:message>
 
   <!-- L2 Common Messages 200 - 399 -->