feat: add partitionId to AsyncSnapshotDirector context

(cherry picked from commit 741da77)

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

@@ -23,6 +23,7 @@
 import io.camunda.zeebe.util.sched.future.CompletableActorFuture;
 import java.time.Duration;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.BooleanSupplier;
 import org.slf4j.Logger;
@@ -57,6 +58,7 @@ public final class AsyncSnapshotDirector extends Actor
   private boolean persistingSnapshot;
   private volatile HealthStatus healthStatus = HealthStatus.HEALTHY;
   private long commitPosition;
+  private final int partitionId;
 
   private AsyncSnapshotDirector(
       final int nodeId,
@@ -69,14 +71,22 @@ private AsyncSnapshotDirector(
     this.stateController = stateController;
     processorName = streamProcessor.getName();
     this.snapshotRate = snapshotRate;
-    actorName = buildActorName(nodeId, "SnapshotDirector", partitionId);
+    this.partitionId = partitionId;
+    actorName = buildActorName(nodeId, "SnapshotDirector", this.partitionId);
     if (streamProcessorMode == StreamProcessorMode.REPLAY) {
       isLastWrittenPositionCommitted = () -> true;
     } else {
       isLastWrittenPositionCommitted = () -> lastWrittenEventPosition <= commitPosition;
     }
   }
 
+  @Override
+  protected Map<String, String> createContext() {
+    final var context = super.createContext();
+    context.put(ACTOR_PROP_PARTITION_ID, Integer.toString(partitionId));
+    return context;
+  }
+
   @Override
   public String getName() {
     return actorName;