Persist snapshot metadata

atomix/cluster/src/test/java/io/atomix/raft/snapshot/InMemorySnapshot.java

@@ -22,6 +22,7 @@
 import io.camunda.zeebe.snapshots.SnapshotChunk;
 import io.camunda.zeebe.snapshots.SnapshotChunkReader;
 import io.camunda.zeebe.snapshots.SnapshotId;
+import io.camunda.zeebe.snapshots.SnapshotMetadata;
 import io.camunda.zeebe.snapshots.SnapshotReservation;
 import io.camunda.zeebe.util.StringUtil;
 import io.camunda.zeebe.util.buffer.BufferUtil;
@@ -149,6 +150,11 @@ public long getChecksum() {
     return 0;
   }
 
+  @Override
+  public SnapshotMetadata getMetadata() {
+    return null;
+  }
+
   @Override
   public ActorFuture<SnapshotReservation> reserve() {
     return null;

broker/src/main/java/io/camunda/zeebe/broker/system/management/BrokerAdminServiceImpl.java

@@ -18,7 +18,7 @@
 import io.camunda.zeebe.scheduler.Actor;
 import io.camunda.zeebe.scheduler.future.ActorFuture;
 import io.camunda.zeebe.snapshots.PersistedSnapshot;
-import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotMetadata;
+import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotId;
 import io.camunda.zeebe.streamprocessor.StreamProcessor;
 import java.util.Collections;
 import java.util.List;
@@ -158,8 +158,8 @@ private void getFollowerPartitionStatus(
     final var snapshotId = getSnapshotId(partition);
     final var processedPositionInSnapshot =
         snapshotId
-            .flatMap(FileBasedSnapshotMetadata::ofFileName)
-            .map(FileBasedSnapshotMetadata::getProcessedPosition)
+            .flatMap(FileBasedSnapshotId::ofFileName)
+            .map(FileBasedSnapshotId::getProcessedPosition)
             .orElse(null);
     final var status =
         PartitionStatus.ofFollower(snapshotId.orElse(null), processedPositionInSnapshot);
@@ -179,8 +179,8 @@ private void getLeaderPartitionStatus(
     final var snapshotId = getSnapshotId(partition);
     final var processedPositionInSnapshot =
         snapshotId
-            .flatMap(FileBasedSnapshotMetadata::ofFileName)
-            .map(FileBasedSnapshotMetadata::getProcessedPosition)
+            .flatMap(FileBasedSnapshotId::ofFileName)
+            .map(FileBasedSnapshotId::getProcessedPosition)
             .orElse(null);
 
     actor.runOnCompletion(

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

@@ -338,7 +338,8 @@ private void persistSnapshotIfLastWrittenPositionCommitted() {
           commitPosition,
           lastWrittenPosition,
           pendingSnapshot);
-      final var snapshotPersistFuture = pendingSnapshot.persist();
+      final var snapshotPersistFuture =
+          pendingSnapshot.withLastFollowupEventPosition(lastWrittenPosition).persist();
 
       snapshotPersistFuture.onComplete(
           (snapshot, persistError) -> {

broker/src/test/java/io/camunda/zeebe/broker/system/partitions/BrokerSnapshotTest.java

@@ -19,7 +19,7 @@
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.ZeebeClientBuilder;
 import io.camunda.zeebe.snapshots.SnapshotId;
-import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotMetadata;
+import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotId;
 import io.netty.util.NetUtil;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
@@ -100,6 +100,6 @@ private SnapshotId waitForSnapshotAtBroker(
                             .getProcessedPositionInSnapshot())
                     .isNotNull());
     final PartitionStatus partitionStatus = brokerAdminService.getPartitionStatus().get(1);
-    return FileBasedSnapshotMetadata.ofFileName(partitionStatus.getSnapshotId()).get();
+    return FileBasedSnapshotId.ofFileName(partitionStatus.getSnapshotId()).get();
   }
 }

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

@@ -19,7 +19,7 @@
 import io.camunda.zeebe.scheduler.testing.ActorSchedulerRule;
 import io.camunda.zeebe.snapshots.PersistedSnapshot;
 import io.camunda.zeebe.snapshots.SnapshotException.StateClosedException;
-import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotMetadata;
+import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotId;
 import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotStore;
 import io.camunda.zeebe.snapshots.impl.SnapshotMetrics;
 import io.camunda.zeebe.test.util.AutoCloseableRule;
@@ -193,9 +193,8 @@ public void shouldTakeSnapshotWhenExporterPositionNotChanged() {
         .extracting(PersistedSnapshot::getCompactionBound)
         .isEqualTo(firstSnapshot.getCompactionBound());
     assertThat(snapshot.getId()).isNotEqualTo(firstSnapshot.getId());
-    final var newSnapshotId = FileBasedSnapshotMetadata.ofFileName(snapshot.getId()).orElseThrow();
-    final var firstSnapshotId =
-        FileBasedSnapshotMetadata.ofFileName(firstSnapshot.getId()).orElseThrow();
+    final var newSnapshotId = FileBasedSnapshotId.ofFileName(snapshot.getId()).orElseThrow();
+    final var firstSnapshotId = FileBasedSnapshotId.ofFileName(firstSnapshot.getId()).orElseThrow();
     assertThat(firstSnapshotId).isLessThan(newSnapshotId);
   }
 
@@ -218,9 +217,8 @@ public void shouldTakeSnapshotWhenProcessorPositionNotChanged() {
         .extracting(PersistedSnapshot::getCompactionBound)
         .isEqualTo(firstSnapshot.getCompactionBound());
     assertThat(snapshot.getId()).isNotEqualTo(firstSnapshot.getId());
-    final var newSnapshotId = FileBasedSnapshotMetadata.ofFileName(snapshot.getId()).orElseThrow();
-    final var firstSnapshotId =
-        FileBasedSnapshotMetadata.ofFileName(firstSnapshot.getId()).orElseThrow();
+    final var newSnapshotId = FileBasedSnapshotId.ofFileName(snapshot.getId()).orElseThrow();
+    final var firstSnapshotId = FileBasedSnapshotId.ofFileName(firstSnapshot.getId()).orElseThrow();
     assertThat(firstSnapshotId).isLessThan(newSnapshotId);
   }
 

qa/integration-tests/src/test/java/io/camunda/zeebe/it/clustering/ClusteringRule.java

@@ -58,7 +58,7 @@
 import io.camunda.zeebe.scheduler.future.ActorFuture;
 import io.camunda.zeebe.scheduler.future.CompletableActorFuture;
 import io.camunda.zeebe.snapshots.SnapshotId;
-import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotMetadata;
+import io.camunda.zeebe.snapshots.impl.FileBasedSnapshotId;
 import io.camunda.zeebe.test.util.AutoCloseableRule;
 import io.camunda.zeebe.test.util.asserts.TopologyAssert;
 import io.camunda.zeebe.test.util.record.RecordingExporterTestWatcher;
@@ -813,7 +813,7 @@ private Optional<SnapshotId> getSnapshot(final Broker broker, final int partitio
 
     return Optional.ofNullable(partitionStatus)
         .map(PartitionStatus::getSnapshotId)
-        .flatMap(FileBasedSnapshotMetadata::ofFileName);
+        .flatMap(FileBasedSnapshotId::ofFileName);
   }
 
   public Optional<SnapshotId> getSnapshot(final int brokerId) {

snapshot/pom.xml

@@ -46,6 +46,16 @@
       <artifactId>slf4j-api</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-annotations</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>io.camunda</groupId>
       <artifactId>zeebe-scheduler</artifactId>

snapshot/src/main/java/io/camunda/zeebe/snapshots/PersistedSnapshot.java

@@ -74,6 +74,12 @@ public interface PersistedSnapshot extends CloseableSilently {
    */
   long getChecksum();
 
+  /**
+   * SnapshotMetadata includes information related to a snapshot.
+   *
+   * @return the metadata of the snapshot.
+   */
+  SnapshotMetadata getMetadata();
   /**
    * Reserves this snapshot. When the snapshot is reserved, it is not deleted until it is released.
    * The reservation status is not persisted. After a restart the snapshot will be in state

snapshot/src/main/java/io/camunda/zeebe/snapshots/SnapshotMetadata.java

@@ -0,0 +1,35 @@
+/*
+ * 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.snapshots;
+
+public interface SnapshotMetadata {
+
+  /**
+   * @return version of the snapshot
+   */
+  int version();
+
+  /**
+   * @return processed position in the snapshot (same as in SnapshotId)
+   */
+  long processedPosition();
+
+  /**
+   * @return exported position in the snapshot (same as in SnapshotId)
+   */
+  long exportedPosition();
+
+  /**
+   * A snapshot is only valid if the logstream consists of the events from the processedPosition up
+   * to the followup event position.
+   *
+   * @return position of the last followUpEvent that must be in the logstream to ensure that the
+   *     system can recover from the snapshot and the logstream.
+   */
+  long lastFollowupEventPosition();
+}

snapshot/src/main/java/io/camunda/zeebe/snapshots/TransientSnapshot.java

@@ -24,4 +24,15 @@ public interface TransientSnapshot extends PersistableSnapshot {
    * @return true on success, false otherwise
    */
   ActorFuture<Void> take(Consumer<Path> takeSnapshot);
+
+  /**
+   * A snapshot is only valid if the accompanying logstream has events from processedPosition up to
+   * the last followup event position. The last followUp event position is the position of an event
+   * whose source position >= actual processed position in the state.
+   *
+   * @param followupEventPosition position of the followup event which must be in the logstream to
+   *     ensure that the system can recover from the snapshot and the events in the logstream.
+   * @return transient snapshot.
+   */
+  TransientSnapshot withLastFollowupEventPosition(long followupEventPosition);
 }

snapshot/src/main/java/io/camunda/zeebe/snapshots/impl/FileBasedReceivedSnapshot.java

@@ -35,16 +35,17 @@ public class FileBasedReceivedSnapshot implements ReceivedSnapshot {
   private final ActorControl actor;
   private final FileBasedSnapshotStore snapshotStore;
 
-  private final FileBasedSnapshotMetadata metadata;
+  private final FileBasedSnapshotId snapshotId;
   private long expectedSnapshotChecksum;
   private int expectedTotalCount;
+  private FileBasedSnapshotMetadata metadata;
 
   FileBasedReceivedSnapshot(
-      final FileBasedSnapshotMetadata metadata,
+      final FileBasedSnapshotId snapshotId,
       final Path directory,
       final FileBasedSnapshotStore snapshotStore,
       final ActorControl actor) {
-    this.metadata = metadata;
+    this.snapshotId = snapshotId;
     this.snapshotStore = snapshotStore;
     this.directory = directory;
     this.actor = actor;
@@ -54,7 +55,7 @@ public class FileBasedReceivedSnapshot implements ReceivedSnapshot {
 
   @Override
   public long index() {
-    return metadata.getIndex();
+    return snapshotId.getIndex();
   }
 
   @Override
@@ -113,6 +114,18 @@ private void applyInternal(final SnapshotChunk snapshotChunk) throws SnapshotWri
 
     LOGGER.trace("Consume snapshot snapshotChunk {} of snapshot {}", chunkName, snapshotId);
     writeReceivedSnapshotChunk(snapshotChunk, snapshotFile);
+
+    if (snapshotChunk.getChunkName().equals(FileBasedSnapshotStore.METADATA_FILE_NAME)) {
+      try {
+        collectMetadata(snapshotChunk.getContent());
+      } catch (final IOException e) {
+        throw new SnapshotWriteException("Cannot decode snapshot metadata");
+      }
+    }
+  }
+
+  private void collectMetadata(final byte[] content) throws IOException {
+    metadata = FileBasedSnapshotMetadata.decode(content);
   }
 
   private void checkChunkChecksumIsValid(
@@ -157,18 +170,18 @@ private void checkTotalCountIsValid(final int currentTotalCount) throws Snapshot
   }
 
   private void checkSnapshotIdIsValid(final String snapshotId) throws SnapshotWriteException {
-    final var receivedSnapshotId = FileBasedSnapshotMetadata.ofFileName(snapshotId);
+    final var receivedSnapshotId = FileBasedSnapshotId.ofFileName(snapshotId);
     if (receivedSnapshotId.isEmpty()) {
       throw new SnapshotWriteException(
           String.format("Snapshot file name '%s' has unexpected format", snapshotId));
     }
 
-    final FileBasedSnapshotMetadata chunkMetadata = receivedSnapshotId.get();
-    if (metadata.compareTo(chunkMetadata) != 0) {
+    final FileBasedSnapshotId chunkSnapshotId = receivedSnapshotId.get();
+    if (this.snapshotId.compareTo(chunkSnapshotId) != 0) {
       throw new SnapshotWriteException(
           String.format(
-              "Expected snapshot chunk metadata to match metadata '%s' but was '%s' instead",
-              metadata, chunkMetadata));
+              "Expected snapshot id in chunk to be '%s' but was '%s' instead",
+              this.snapshotId, chunkSnapshotId));
     }
   }
 
@@ -213,7 +226,7 @@ public ActorFuture<PersistedSnapshot> persist() {
 
   @Override
   public SnapshotId snapshotId() {
-    return metadata;
+    return snapshotId;
   }
 
   @Override
@@ -233,7 +246,7 @@ private void abortInternal() {
   }
 
   private void persistInternal(final CompletableActorFuture<PersistedSnapshot> future) {
-    if (snapshotStore.hasSnapshotId(metadata.getSnapshotIdAsString())) {
+    if (snapshotStore.hasSnapshotId(snapshotId.getSnapshotIdAsString())) {
       abortInternal();
       future.complete(snapshotStore.getLatestSnapshot().orElseThrow());
       return;
@@ -257,8 +270,17 @@ private void persistInternal(final CompletableActorFuture<PersistedSnapshot> fut
     }
 
     try {
+      if (metadata == null) {
+        // backward compatibility
+        metadata =
+            new FileBasedSnapshotMetadata(
+                FileBasedSnapshotStore.VERSION,
+                snapshotId.getProcessedPosition(),
+                snapshotId.getExportedPosition(),
+                Long.MAX_VALUE);
+      }
       final PersistedSnapshot value =
-          snapshotStore.newSnapshot(metadata, directory, expectedSnapshotChecksum);
+          snapshotStore.newSnapshot(snapshotId, directory, expectedSnapshotChecksum, metadata);
       future.complete(value);
     } catch (final Exception e) {
       future.completeExceptionally(e);
@@ -275,7 +297,7 @@ public String toString() {
         + ", snapshotStore="
         + snapshotStore.getName()
         + ", metadata="
-        + metadata
+        + snapshotId
         + '}';
   }
 }