fix(protocol): fix random generation of CommandDistributionRecords

The CommandDistributionRecord contains a ValueType and a UnifiedRecordValue. We must make sure during the random generation of this record that the UnifiedRecordValue matches the ValueType. This is achieved by creating a custom randomizer for this specific class.

A downside of this implementation is that the ProtocolFactory is now aware of this specific Record. When we make changes in the Record we must change them in this randomizer as well.

Author: remcowesterhoud

protocol-impl/src/main/java/io/camunda/zeebe/protocol/impl/record/value/distribution/CommandDistributionRecord.java

@@ -34,6 +34,12 @@ public final class CommandDistributionRecord extends UnifiedRecordValue
     RECORDS_BY_TYPE.put(ValueType.DEPLOYMENT, DeploymentRecord::new);
   }
 
+  /*
+   NOTE! When adding a new property here it must also be added to the ProtocolFactory! This class
+   contains a randomizer implementation which is used to generate a random
+   CommandDistributionRecord. The new property must be added there. Without it we won't generate a
+   complete record.
+  */
   private final IntegerProperty partitionIdProperty = new IntegerProperty("partitionId");
   private final EnumProperty<ValueType> valueTypeProperty =
       new EnumProperty<>("valueType", ValueType.class);

protocol-jackson/src/main/java/io/camunda/zeebe/protocol/jackson/RecordMixin.java

@@ -34,4 +34,8 @@ abstract class RecordMixin<T extends RecordValue> {
   @JsonTypeInfo(use = Id.CUSTOM, include = As.EXTERNAL_PROPERTY, property = "valueType")
   @JsonTypeIdResolver(RecordValueTypeIdResolver.class)
   private T value;
+
+  @JsonTypeInfo(use = Id.CUSTOM, include = As.EXTERNAL_PROPERTY, property = "valueType")
+  @JsonTypeIdResolver(RecordValueTypeIdResolver.class)
+  private T commandValue;
 }

protocol-jackson/src/main/java/io/camunda/zeebe/protocol/jackson/ZeebeProtocolModule.java

@@ -11,6 +11,7 @@
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import io.camunda.zeebe.protocol.record.ImmutableRecord;
 import io.camunda.zeebe.protocol.record.Record;
+import io.camunda.zeebe.protocol.record.value.ImmutableCommandDistributionRecordValue;
 
 /**
  * A Jackson module which enables your {@link ObjectMapper} to serialize and deserialize Zeebe
@@ -34,6 +35,7 @@
 public final class ZeebeProtocolModule extends SimpleModule {
   public ZeebeProtocolModule() {
     setMixInAnnotation(ImmutableRecord.Builder.class, RecordMixin.class);
+    setMixInAnnotation(ImmutableCommandDistributionRecordValue.Builder.class, RecordMixin.class);
   }
 
   @Override

protocol-test-util/src/main/java/io/camunda/zeebe/test/broker/protocol/ProtocolFactory.java

@@ -15,6 +15,7 @@
 import io.camunda.zeebe.protocol.record.RecordValue;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.ValueTypeMapping;
+import io.camunda.zeebe.protocol.record.value.ImmutableCommandDistributionRecordValue;
 import io.github.classgraph.ClassGraph;
 import io.github.classgraph.ClassInfo;
 import io.github.classgraph.ClassInfoList;
@@ -231,6 +232,18 @@ private void registerRandomizers() {
     final var recordTypes = EnumSet.complementOf(excludedRecordTypes);
     randomizerRegistry.registerRandomizer(
         RecordType.class, new EnumRandomizer<>(getSeed(), recordTypes.toArray(RecordType[]::new)));
+
+    randomizerRegistry.registerRandomizer(
+        ImmutableCommandDistributionRecordValue.class,
+        () -> {
+          final var valueType = random.nextObject(ValueType.class);
+          final var typeInfo = ValueTypeMapping.get(valueType);
+          return ImmutableCommandDistributionRecordValue.builder()
+              .withPartitionId(random.nextInt())
+              .withValueType(valueType)
+              .withCommandValue(generateObject(typeInfo.getValueClass()))
+              .build();
+        });
   }
 
   private void registerProtocolType(final ClassInfo abstractType) {