merge: #9635

9635: [Backport stable/1.3] fix(broker): do not log error if follower fails to take snapshot when log is not uptodate r=deepthidevaki a=backport-action

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

relates to #7911

Co-authored-by: Deepthi Devaki Akkoorath <deepthidevaki@gmail.com>

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/NoEntryAtSnapshotPosition.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.broker.system.partitions;
+
+/**
+ * Used when there is no entry at the determined snapshot position while taking a transient
+ * snapshot.
+ */
+public class NoEntryAtSnapshotPosition extends RuntimeException {
+
+  public NoEntryAtSnapshotPosition(final String message) {
+    super(message);
+  }
+}

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/impl/AsyncSnapshotDirector.java

@@ -9,6 +9,7 @@
 
 import io.atomix.raft.RaftCommittedEntryListener;
 import io.atomix.raft.storage.log.IndexedRaftLogEntry;
+import io.camunda.zeebe.broker.system.partitions.NoEntryAtSnapshotPosition;
 import io.camunda.zeebe.broker.system.partitions.StateController;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessor;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorMode;
@@ -51,6 +52,7 @@ public final class AsyncSnapshotDirector extends Actor
   private final String processorName;
   private final StreamProcessor streamProcessor;
   private final String actorName;
+  private final StreamProcessorMode streamProcessorMode;
   private final Set<FailureListener> listeners = new HashSet<>();
   private final BooleanSupplier isLastWrittenPositionCommitted;
 
@@ -79,6 +81,7 @@ private AsyncSnapshotDirector(
     this.snapshotRate = snapshotRate;
     this.partitionId = partitionId;
     actorName = buildActorName(nodeId, "SnapshotDirector", this.partitionId);
+    this.streamProcessorMode = streamProcessorMode;
     if (streamProcessorMode == StreamProcessorMode.REPLAY) {
       isLastWrittenPositionCommitted = () -> true;
     } else {
@@ -256,11 +259,7 @@ private void takeSnapshot() {
     transientSnapshotFuture.onComplete(
         (transientSnapshot, snapshotTakenError) -> {
           if (snapshotTakenError != null) {
-            if (snapshotTakenError instanceof SnapshotException.SnapshotAlreadyExistsException) {
-              LOG.debug("Did not take a snapshot. {}", snapshotTakenError.getMessage());
-            } else {
-              LOG.error("Failed to take a snapshot for {}", processorName, snapshotTakenError);
-            }
+            logSnapshotTakenError(snapshotTakenError);
             resetStateOnFailure(snapshotTakenError);
             return;
           }
@@ -269,6 +268,20 @@ private void takeSnapshot() {
         });
   }
 
+  private void logSnapshotTakenError(final Throwable snapshotTakenError) {
+    if (snapshotTakenError instanceof SnapshotException.SnapshotAlreadyExistsException) {
+      LOG.debug("Did not take a snapshot. {}", snapshotTakenError.getMessage());
+    } else if (snapshotTakenError instanceof NoEntryAtSnapshotPosition
+        && streamProcessorMode == StreamProcessorMode.REPLAY) {
+      LOG.debug(
+          "Did not take a snapshot: {}. Most likely this partition has not received the entry yet. Will retry in {}",
+          snapshotTakenError.getMessage(),
+          snapshotRate);
+    } else {
+      LOG.error("Failed to take a snapshot for {}", processorName, snapshotTakenError);
+    }
+  }
+
   private void onTransientSnapshotTaken(final TransientSnapshot transientSnapshot) {
 
     pendingSnapshot = transientSnapshot;

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/impl/StateControllerImpl.java

@@ -8,6 +8,7 @@
 package io.camunda.zeebe.broker.system.partitions.impl;
 
 import io.camunda.zeebe.broker.system.partitions.AtomixRecordEntrySupplier;
+import io.camunda.zeebe.broker.system.partitions.NoEntryAtSnapshotPosition;
 import io.camunda.zeebe.broker.system.partitions.StateController;
 import io.camunda.zeebe.db.ZeebeDb;
 import io.camunda.zeebe.db.ZeebeDbFactory;
@@ -155,7 +156,7 @@ private void takeTransientSnapshotInternal(
 
       if (optionalIndexed.isEmpty()) {
         future.completeExceptionally(
-            new IllegalStateException(
+            new NoEntryAtSnapshotPosition(
                 String.format(
                     "Failed to take snapshot. Expected to find an indexed entry for determined snapshot position %d (processedPosition = %d, exportedPosition=%d), but found no matching indexed entry which contains this position.",
                     snapshotPosition, lowerBoundSnapshotPosition, exportedPosition)));

broker/src/test/java/io/camunda/zeebe/broker/system/partitions/impl/StateControllerImplTest.java

@@ -12,6 +12,8 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.atomix.raft.storage.log.entry.ApplicationEntry;
+import io.camunda.zeebe.broker.system.partitions.AtomixRecordEntrySupplier;
+import io.camunda.zeebe.broker.system.partitions.NoEntryAtSnapshotPosition;
 import io.camunda.zeebe.broker.system.partitions.TestIndexedRaftLogEntry;
 import io.camunda.zeebe.db.impl.rocksdb.ZeebeRocksDbFactory;
 import io.camunda.zeebe.snapshots.PersistedSnapshot;
@@ -28,6 +30,7 @@
 import java.nio.file.StandardOpenOption;
 import java.util.Comparator;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
 import org.agrona.collections.MutableLong;
 import org.agrona.concurrent.UnsafeBuffer;
 import org.junit.Before;
@@ -46,6 +49,14 @@ public final class StateControllerImplTest {
   private FileBasedSnapshotStore store;
   private Path runtimeDirectory;
 
+  private final AtomixRecordEntrySupplier indexedRaftLogEntry =
+      l ->
+          Optional.of(
+              new TestIndexedRaftLogEntry(l, 1, new ApplicationEntry(1, 10, new UnsafeBuffer())));
+  private final AtomixRecordEntrySupplier emptyEntrySupplier = l -> Optional.empty();
+  private final AtomicReference<AtomixRecordEntrySupplier> atomixRecordEntrySupplier =
+      new AtomicReference<>(indexedRaftLogEntry);
+
   @Before
   public void setup() throws IOException {
 
@@ -64,10 +75,7 @@ public void setup() throws IOException {
             ZeebeRocksDbFactory.newFactory(),
             store,
             runtimeDirectory,
-            l ->
-                Optional.of(
-                    new TestIndexedRaftLogEntry(
-                        l, 1, new ApplicationEntry(1, 10, new UnsafeBuffer()))),
+            l -> atomixRecordEntrySupplier.get().getPreviousIndexedEntry(l),
             db -> exporterPosition.get(),
             store);
 
@@ -84,6 +92,17 @@ public void shouldNotTakeSnapshotIfDbIsClosed() {
         .hasCauseInstanceOf(StateClosedException.class);
   }
 
+  @Test
+  public void shouldNotTakeSnapshotIfNoIndexedEntry() {
+    // given
+    atomixRecordEntrySupplier.set(emptyEntrySupplier);
+    snapshotController.recover().join();
+
+    // then
+    assertThatThrownBy(() -> snapshotController.takeTransientSnapshot(1).join())
+        .hasCauseInstanceOf(NoEntryAtSnapshotPosition.class);
+  }
+
   @Test
   public void shouldTakeTempSnapshotWithExporterPosition() {
     // given