fix(engine): update the key generator on replay

On replay, update the key generator with the replayed record's key. This
fixes an issue where the follower would restart with a snapshot where
nothing needed to be replayed, and would thus not update its generated
key, leading to processing starting with the wrong key and duplicate
keys, causing state inconsistencies.

engine/src/main/java/io/camunda/zeebe/engine/processing/streamprocessor/ReplayStateMachine.java

@@ -73,7 +73,6 @@ public final class ReplayStateMachine implements LogRecordAwaiter {
   private long batchSourceEventPosition = StreamProcessor.UNSET_POSITION;
 
   private long snapshotPosition;
-  private long highestRecordKey = -1L;
   private long lastReadRecordPosition = StreamProcessor.UNSET_POSITION;
   private long lastReplayedEventPosition = StreamProcessor.UNSET_POSITION;
 
@@ -234,9 +233,6 @@ private void replayEvent(final LoggedEvent currentEvent) {
    * the last source event, which has caused the last applied event.
    */
   private void onRecordsReplayed() {
-    // restore the key generate with the highest key from the log
-    keyGeneratorControls.setKeyIfHigher(highestRecordKey);
-
     LOG.info(LOG_STMT_REPLAY_FINISHED, lastReadRecordPosition);
     recoveryFuture.complete(lastSourceEventPosition);
   }
@@ -267,8 +263,7 @@ private void onRecordReplayed(final LoggedEvent currentEvent) {
 
     // records from other partitions should not influence the key generator of this partition
     if (Protocol.decodePartitionId(currentRecordKey) == zeebeState.getPartitionId()) {
-      // remember the highest key on the stream to restore the key generator after replay
-      highestRecordKey = Math.max(currentRecordKey, highestRecordKey);
+      keyGeneratorControls.setKeyIfHigher(currentRecordKey);
     }
   }