merge: #8994

8994: Fix notification of snapshot replication listeners about missed events r=oleschoenburg a=oleschoenburg

## Description
Whenever a raft partition transitions to a new role, we must reset `missedSnapshotReplicationEvents` so that registering new snapshot replication listeners does not trigger the listener for snapshot replication events that occurred for a different role.

This is to guard against a known case where raft received a snapshot and transitioned to leader before role change and snapshot replication listeners were registered. Once the registration of the snapshot replication listener registration went through, the listener was informed about the missed replication and caused the zeebe partition to transition to follower, while the raft partition was leader.

Additionally, because we keep track of missed snapshot replication events to notify listeners on registering, the partition is not necessarily in follower role at that point. This breaks the assumption of the listener, so we add a condition here to only trigger listeners on register if the raft partition is still follower.

## Related issues

relates to #8830
closes #8978


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

atomix/cluster/src/main/java/io/atomix/raft/impl/RaftContext.java

@@ -476,13 +476,15 @@ public void addSnapshotReplicationListener(
           snapshotReplicationListeners.add(snapshotReplicationListener);
           // Notify listener immediately if it registered during an ongoing replication.
           // This is to prevent missing necessary state transitions.
-          switch (missedSnapshotReplicationEvents) {
-            case STARTED -> snapshotReplicationListener.onSnapshotReplicationStarted();
-            case COMPLETED -> {
-              snapshotReplicationListener.onSnapshotReplicationStarted();
-              snapshotReplicationListener.onSnapshotReplicationCompleted(term);
+          if (role.role() == Role.FOLLOWER) {
+            switch (missedSnapshotReplicationEvents) {
+              case STARTED -> snapshotReplicationListener.onSnapshotReplicationStarted();
+              case COMPLETED -> {
+                snapshotReplicationListener.onSnapshotReplicationStarted();
+                snapshotReplicationListener.onSnapshotReplicationCompleted(term);
+              }
+              default -> {}
             }
-            default -> {}
           }
         });
   }
@@ -676,6 +678,7 @@ private void startTransition() {
   }
 
   private void completeTransition() {
+    missedSnapshotReplicationEvents = MissedSnapshotReplicationEvents.NONE;
     ongoingTransition = false;
   }
 

atomix/cluster/src/test/java/io/atomix/raft/SnapshotReplicationListenerTest.java

@@ -100,4 +100,18 @@ public void shouldNotCallListenerOnRegister() {
     verify(snapshotReplicationListener, times(0))
         .onSnapshotReplicationCompleted(follower.getTerm());
   }
+
+  @Test
+  public void shouldNotCallListenerOnRegisterIfLeader() {
+    // given
+    final var snapshotReplicationListener = mock(SnapshotReplicationListener.class);
+    final var leader = raftRule.getLeader().orElseThrow();
+    // when
+    leader.getContext().notifySnapshotReplicationStarted();
+    leader.getContext().notifySnapshotReplicationCompleted();
+    leader.getContext().addSnapshotReplicationListener(snapshotReplicationListener);
+    // then
+    verify(snapshotReplicationListener, times(0)).onSnapshotReplicationStarted();
+    verify(snapshotReplicationListener, times(0)).onSnapshotReplicationCompleted(leader.getTerm());
+  }
 }