merge: #8568

8568: [Backport stable/1.3] Configurable threshold for preferring snapshot replication r=oleschoenburg a=github-actions[bot]

# Description
Backport of #8562 to `stable/1.3`.

relates to #7968

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

atomix/cluster/src/main/java/io/atomix/raft/partition/RaftPartitionGroup.java

@@ -492,6 +492,22 @@ public Builder withPartitionDistributor(final PartitionDistributor partitionDist
       return this;
     }
 
+    /**
+     * Sets the threshold for preferring snapshot replication. The unit is <i>number of records</i>
+     * by which a follower may lag behind before the leader starts to prefer replicating snapshots
+     * instead of records.
+     *
+     * @param preferSnapshotReplicationThreshold the threshold to use
+     * @return this builder for chaining
+     */
+    public Builder withPreferSnapshotReplicationThreshold(
+        final int preferSnapshotReplicationThreshold) {
+      config
+          .getPartitionConfig()
+          .setPreferSnapshotReplicationThreshold(preferSnapshotReplicationThreshold);
+      return this;
+    }
+
     @Override
     public RaftPartitionGroup build() {
       return new RaftPartitionGroup(config);

broker/src/main/java/io/camunda/zeebe/broker/partitioning/RaftPartitionGroupFactory.java

@@ -75,7 +75,9 @@ RaftPartitionGroup buildRaftPartitionGroup(
             .withHeartbeatInterval(clusterCfg.getHeartbeatInterval())
             .withRequestTimeout(experimentalCfg.getRaft().getRequestTimeout())
             .withMaxQuorumResponseTimeout(experimentalCfg.getRaft().getMaxQuorumResponseTimeout())
-            .withMinStepDownFailureCount(experimentalCfg.getRaft().getMinStepDownFailureCount());
+            .withMinStepDownFailureCount(experimentalCfg.getRaft().getMinStepDownFailureCount())
+            .withPreferSnapshotReplicationThreshold(
+                experimentalCfg.getRaft().getPreferSnapshotReplicationThreshold());
 
     final int maxMessageSize = (int) networkCfg.getMaxMessageSizeInBytes();
 

broker/src/main/java/io/camunda/zeebe/broker/system/configuration/ExperimentalRaftCfg.java

@@ -14,10 +14,12 @@ public final class ExperimentalRaftCfg implements ConfigurationEntry {
   private static final Duration DEFAULT_REQUEST_TIMEOUT = Duration.ofSeconds(5);
   private static final Duration DEFAULT_MAX_QUORUM_RESPONSE_TIMEOUT = Duration.ofSeconds(0);
   private static final int DEFAULT_MIN_STEP_DOWN_FAILURE_COUNT = 3;
+  private static final int DEFAULT_PREFER_SNAPSHOT_REPLICATION_THRESHOLD = 100;
 
   private Duration requestTimeout = DEFAULT_REQUEST_TIMEOUT;
   private Duration maxQuorumResponseTimeout = DEFAULT_MAX_QUORUM_RESPONSE_TIMEOUT;
   private int minStepDownFailureCount = DEFAULT_MIN_STEP_DOWN_FAILURE_COUNT;
+  private int preferSnapshotReplicationThreshold = DEFAULT_PREFER_SNAPSHOT_REPLICATION_THRESHOLD;
 
   public Duration getRequestTimeout() {
     return requestTimeout;
@@ -42,4 +44,12 @@ public int getMinStepDownFailureCount() {
   public void setMinStepDownFailureCount(final int minStepDownFailureCount) {
     this.minStepDownFailureCount = minStepDownFailureCount;
   }
+
+  public int getPreferSnapshotReplicationThreshold() {
+    return preferSnapshotReplicationThreshold;
+  }
+
+  public void setPreferSnapshotReplicationThreshold(final int preferSnapshotReplicationThreshold) {
+    this.preferSnapshotReplicationThreshold = preferSnapshotReplicationThreshold;
+  }
 }

broker/src/test/java/io/camunda/zeebe/broker/partitioning/RaftPartitionGroupFactoryTest.java

@@ -141,6 +141,18 @@ void shouldDisablePriorityElection() {
     assertThat(config.getPartitionConfig().isPriorityElectionEnabled()).isFalse();
   }
 
+  @Test
+  void shouldSetPreferSnapshotReplicationThreshold() {
+    // given
+    brokerCfg.getExperimental().getRaft().setPreferSnapshotReplicationThreshold(1000);
+
+    // when
+    final var config = buildRaftPartitionGroup();
+
+    // then
+    assertThat(config.getPartitionConfig().getPreferSnapshotReplicationThreshold()).isEqualTo(1000);
+  }
+
   private RaftPartitionGroupConfig buildRaftPartitionGroup() {
     final var partitionGroup = factory.buildRaftPartitionGroup(brokerCfg, SNAPSHOT_STORE_FACTORY);
     return (RaftPartitionGroupConfig) partitionGroup.config();

broker/src/test/java/io/camunda/zeebe/broker/system/configuration/ExperimentalCfgTest.java

@@ -86,4 +86,27 @@ public void shouldSetRaftMinStepDownFailureCountFromEnv() {
     // then
     assertThat(raft.getMinStepDownFailureCount()).isEqualTo(10);
   }
+
+  @Test
+  public void shouldSetPreferSnapshotReplicationThresholdFromConfig() {
+    // when
+    final BrokerCfg cfg = TestConfigReader.readConfig("experimental-cfg", environment);
+    final var raft = cfg.getExperimental().getRaft();
+
+    // then
+    assertThat(raft.getPreferSnapshotReplicationThreshold()).isEqualTo(500);
+  }
+
+  @Test
+  public void shouldSetPreferSnapshotReplicationThresholdFromEnv() {
+    // given
+    environment.put("zeebe.broker.experimental.raft.preferSnapshotReplicationThreshold", "10");
+
+    // when
+    final BrokerCfg cfg = TestConfigReader.readConfig("experimental-cfg", environment);
+    final var raft = cfg.getExperimental().getRaft();
+
+    // then
+    assertThat(raft.getPreferSnapshotReplicationThreshold()).isEqualTo(10);
+  }
 }

broker/src/test/resources/system/experimental-cfg.yaml

@@ -6,5 +6,6 @@ zeebe:
         requestTimeout: 10s
         maxQuorumResponseTimeout: 8s
         minStepDownFailureCount: 5
+        preferSnapshotReplicationThreshold: 500
       queryApi:
         enabled: true

dist/src/main/config/broker.standalone.yaml.template

@@ -620,6 +620,12 @@
         # This setting can also be overridden using the environment variable ZEEBE_BROKER_EXPERIMENTAL_RAFT_MAXQUORUMRESPONSETIMEOUT
         # maxQuorumResponseTimeout = 0ms
 
+        # Threshold used by the leader to decide between replicating a snapshot or records.
+        # The unit is number of records by which the follower may lag behind before the leader
+        # prefers replicating snapshots instead of records.
+        # This setting can also be overridden using the environment variable ZEEBE_BROKER_EXPERIMENTAL_RAFT_PREFERSNAPSHOTREPLICATIONTHRESHOLD.
+        # preferSnapshotReplicationThreshold = 100
+
       # Allows to configure RocksDB properties, which is used for state management.
       # rocksdb:
         # Specify custom column family options overwriting Zeebe's own defaults.

dist/src/main/config/broker.yaml.template

@@ -558,6 +558,12 @@
         # This setting can also be overridden using the environment variable ZEEBE_BROKER_EXPERIMENTAL_RAFT_MAXQUORUMRESPONSETIMEOUT
         # maxQuorumResponseTimeout = 0ms
 
+        # Threshold used by the leader to decide between replicating a snapshot or records.
+        # The unit is number of records by which the follower may lag behind before the leader
+        # prefers replicating snapshots instead of records.
+        # This setting can also be overridden using the environment variable ZEEBE_BROKER_EXPERIMENTAL_RAFT_PREFERSNAPSHOTREPLICATIONTHRESHOLD.
+        # preferSnapshotReplicationThreshold = 100
+
       # Allows to configure RocksDB properties, which is used for state management.
       # rocksdb:
         # Specify custom column family options overwriting Zeebe's own defaults.