Snapshotting

pom.xml

@@ -44,7 +44,7 @@
         <sonar.organization>axoniq</sonar.organization>
         <sonar.host.url>https://sonarcloud.io</sonar.host.url>
 
-        <axonserver.api.version>2025.1.2</axonserver.api.version>
+        <axonserver.api.version>2025.1.2-SNAPSHOT</axonserver.api.version>
 
         <grpc.version>1.75.0</grpc.version>
         <protobuf.version>4.32.1</protobuf.version>

src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java

@@ -21,6 +21,7 @@
 import io.axoniq.axonserver.connector.control.ControlChannel;
 import io.axoniq.axonserver.connector.event.DcbEventChannel;
 import io.axoniq.axonserver.connector.event.EventChannel;
+import io.axoniq.axonserver.connector.event.SnapshotChannel;
 import io.axoniq.axonserver.connector.event.transformation.EventTransformationChannel;
 import io.axoniq.axonserver.connector.query.QueryChannel;
 
@@ -96,6 +97,13 @@ public interface AxonServerConnection {
      */
     QueryChannel queryChannel();
 
+    /**
+     * Returns the channel on which Snapshotting can be performed with AxonServer.
+     *
+     * @return the channel for Snapshotting
+     */
+    SnapshotChannel snapshotChannel();
+
     /**
      * Returns the channel on which Event transformations can be performed with AxonServer.
      *

src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java

@@ -0,0 +1,59 @@
+package io.axoniq.axonserver.connector.event;
+
+import io.axoniq.axonserver.connector.ResultStream;
+import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotRequest;
+import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotResponse;
+import io.axoniq.axonserver.grpc.event.dcb.DeleteSnapshotsRequest;
+import io.axoniq.axonserver.grpc.event.dcb.DeleteSnapshotsResponse;
+import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotRequest;
+import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotResponse;
+import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsRequest;
+import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsResponse;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Communication channel for Snapshot-related interactions with AxonServer. This interface provides operations for
+ * managing snapshots, including adding, deleting, listing, and retrieving snapshots.
+ */
+public interface SnapshotChannel {
+
+    /**
+     * Adds a snapshot to the store. If a snapshot with the same key and sequence already exists, it will be replaced.
+     * When the prune flag is set to true, all snapshots with the same key and a sequence number less than the provided
+     * sequence will be removed.
+     *
+     * @param request The request containing the snapshot to add, including key, sequence, prune flag, and snapshot
+     *                data
+     * @return A CompletableFuture that resolves to an AddSnapshotResponse when the operation completes
+     */
+    CompletableFuture<AddSnapshotResponse> addSnapshot(AddSnapshotRequest request);
+
+    /**
+     * Deletes snapshots within a specified sequence range for a given key. The range is inclusive at the lower bound
+     * (fromSequence) and exclusive at the upper bound (toSequence).
+     *
+     * @param request The request containing the key and sequence range to delete
+     * @return A CompletableFuture that resolves to a DeleteSnapshotsResponse when the operation completes
+     */
+    CompletableFuture<DeleteSnapshotsResponse> deleteSnapshots(DeleteSnapshotsRequest request);
+
+    /**
+     * Lists snapshots for a given key within a specified sequence range. The range is inclusive at the lower bound
+     * (fromSequence) and exclusive at the upper bound (toSequence). Snapshots are returned in ascending order by
+     * sequence number.
+     *
+     * @param request The request containing the key and sequence range to list
+     * @return A ResultStream of ListSnapshotsResponse objects containing the matching snapshots
+     */
+    ResultStream<ListSnapshotsResponse> listSnapshots(ListSnapshotsRequest request);
+
+    /**
+     * Retrieves the snapshot with the highest sequence number for a given key. If no snapshots exist for the key, the
+     * response will not contain a snapshot.
+     *
+     * @param request The request containing the key to get the last snapshot for
+     * @return A CompletableFuture that resolves to a GetLastSnapshotResponse when the operation completes
+     */
+    CompletableFuture<GetLastSnapshotResponse> getLastSnapshot(GetLastSnapshotRequest request);
+}

src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java

@@ -0,0 +1,131 @@
+package io.axoniq.axonserver.connector.event.impl;
+
+import com.google.protobuf.Empty;
+import io.axoniq.axonserver.connector.ResultStream;
+import io.axoniq.axonserver.connector.event.SnapshotChannel;
+import io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel;
+import io.axoniq.axonserver.connector.impl.AbstractBufferedStream;
+import io.axoniq.axonserver.connector.impl.AxonServerManagedChannel;
+import io.axoniq.axonserver.connector.impl.FutureStreamObserver;
+import io.axoniq.axonserver.grpc.FlowControl;
+import io.axoniq.axonserver.grpc.control.ClientIdentification;
+import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotRequest;
+import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotResponse;
+import io.axoniq.axonserver.grpc.event.dcb.DcbSnapshotStoreGrpc;
+import io.axoniq.axonserver.grpc.event.dcb.DeleteSnapshotsRequest;
+import io.axoniq.axonserver.grpc.event.dcb.DeleteSnapshotsResponse;
+import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotRequest;
+import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotResponse;
+import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsRequest;
+import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsResponse;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+
+/**
+ * {@link SnapshotChannel} implementation, serving as the event connection between Axon Server and a client
+ * application.
+ *
+ * @author Silvano Biemans
+ * @author Milan Savic
+ * @since 2025.2.0
+ */
+public class SnapshotChannelImpl extends AbstractAxonServerChannel<Void> implements SnapshotChannel {
+
+    private final DcbSnapshotStoreGrpc.DcbSnapshotStoreStub snapshotStore;
+    private final ClientIdentification clientIdentification;
+    private final Set<ResultStream<ListSnapshotsResponse>> buffers = ConcurrentHashMap.newKeySet();
+
+    private static final int BUFFER_SIZE = 512;
+    private static final int REFILL_BATCH = 16;
+
+    /**
+     * Instantiate an {@link AbstractAxonServerChannel}.
+     *
+     * @param clientIdentification     identifies a client
+     * @param executor                 a {@link ScheduledExecutorService} used to schedule reconnections
+     * @param axonServerManagedChannel the {@link AxonServerManagedChannel} used to connect to AxonServer
+     */
+    public SnapshotChannelImpl(ClientIdentification clientIdentification, ScheduledExecutorService executor,
+                               AxonServerManagedChannel axonServerManagedChannel) {
+        super(clientIdentification, executor, axonServerManagedChannel);
+        this.snapshotStore = DcbSnapshotStoreGrpc.newStub(axonServerManagedChannel);
+        this.clientIdentification = clientIdentification;
+    }
+
+    @Override
+    public CompletableFuture<AddSnapshotResponse> addSnapshot(AddSnapshotRequest request) {
+        FutureStreamObserver<AddSnapshotResponse> res = new FutureStreamObserver<>(null);
+        this.snapshotStore.add(request, res);
+        return res;
+    }
+
+    @Override
+    public CompletableFuture<DeleteSnapshotsResponse> deleteSnapshots(DeleteSnapshotsRequest request) {
+        FutureStreamObserver<DeleteSnapshotsResponse> res = new FutureStreamObserver<>(null);
+        this.snapshotStore.delete(request, res);
+        return res;
+    }
+
+    @Override
+    public ResultStream<ListSnapshotsResponse> listSnapshots(ListSnapshotsRequest request) {
+        AbstractBufferedStream<ListSnapshotsResponse, Empty> buffer =
+                new AbstractBufferedStream<>(clientIdentification.getClientId(),
+                                             BUFFER_SIZE,
+                                             REFILL_BATCH) {
+                    @Override
+                    protected ListSnapshotsResponse terminalMessage() {
+                        return ListSnapshotsResponse.newBuilder()
+                                                    .build();
+                    }
+
+                    @Override
+                    protected Empty buildFlowControlMessage(FlowControl flowControl) {
+                        return null;
+                    }
+                };
+        buffers.add(buffer);
+        buffer.onCloseRequested(() -> buffers.remove(buffer));
+        try {
+            snapshotStore.list(request, buffer);
+        } catch (Exception e) {
+            buffers.remove(buffer);
+            throw e;
+        }
+        return buffer;
+    }
+
+    @Override
+    public CompletableFuture<GetLastSnapshotResponse> getLastSnapshot(GetLastSnapshotRequest request) {
+        FutureStreamObserver<GetLastSnapshotResponse> res = new FutureStreamObserver<>(null);
+        this.snapshotStore.getLast(request, res);
+        return res;
+    }
+
+    @Override
+    public void connect() {
+        // nothing to do here
+    }
+
+    @Override
+    public void reconnect() {
+        closeBuffers();
+    }
+
+    @Override
+    public void disconnect() {
+        closeBuffers();
+    }
+
+    @Override
+    public boolean isReady() {
+        return true;
+    }
+
+    private void closeBuffers() {
+        buffers.forEach(ResultStream::close);
+        buffers.clear();
+    }
+}

src/main/java/io/axoniq/axonserver/connector/impl/ContextConnection.java

@@ -24,6 +24,8 @@
 import io.axoniq.axonserver.connector.control.ControlChannel;
 import io.axoniq.axonserver.connector.event.DcbEventChannel;
 import io.axoniq.axonserver.connector.event.EventChannel;
+import io.axoniq.axonserver.connector.event.SnapshotChannel;
+import io.axoniq.axonserver.connector.event.impl.SnapshotChannelImpl;
 import io.axoniq.axonserver.connector.event.impl.DcbEventChannelImpl;
 import io.axoniq.axonserver.connector.event.impl.EventChannelImpl;
 import io.axoniq.axonserver.connector.event.transformation.EventTransformationChannel;
@@ -58,6 +60,7 @@ public class ContextConnection implements AxonServerConnection {
     private final AtomicReference<QueryChannelImpl> queryChannel = new AtomicReference<>();
     private final AtomicReference<EventTransformationChannelImpl> eventTransformationChannel = new AtomicReference<>();
     private final AtomicReference<AdminChannelImpl> adminChannel = new AtomicReference<>();
+    private final AtomicReference<SnapshotChannelImpl> snapshotChannel = new AtomicReference<>();
     private final ScheduledExecutorService executorService;
     private final AxonServerManagedChannel connection;
     private final int commandPermits;
@@ -112,6 +115,7 @@ private void reconnectChannels() {
         doIfNotNull(eventChannel.get(), EventChannelImpl::reconnect);
         doIfNotNull(eventTransformationChannel.get(), EventTransformationChannelImpl::reconnect);
         doIfNotNull(adminChannel.get(), AdminChannelImpl::reconnect);
+        doIfNotNull(snapshotChannel.get(), SnapshotChannelImpl::reconnect);
     }
 
     @Override
@@ -144,6 +148,7 @@ public void disconnect() {
         doIfNotNull(eventChannel.get(), EventChannelImpl::disconnect);
         doIfNotNull(eventTransformationChannel.get(), EventTransformationChannelImpl::disconnect);
         doIfNotNull(adminChannel.get(), AdminChannelImpl::disconnect);
+        doIfNotNull(snapshotChannel.get(), SnapshotChannelImpl::disconnect);
         connection.shutdown();
         onShutdown.accept(this);
         try {
@@ -190,6 +195,16 @@ public EventChannel eventChannel() {
         );
     }
 
+    @Override
+    public SnapshotChannel snapshotChannel() {
+        return createIfAbsentAndInitialize(snapshotChannel,
+                                           () -> new SnapshotChannelImpl(clientIdentification,
+                                                                         executorService,
+                                                                         connection),
+                                           this::ensureConnected
+                );
+    }
+
     @Override
     public DcbEventChannel dcbEventChannel() {
         return createIfAbsentAndInitialize(dcbEventChannel,

src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java

@@ -38,7 +38,9 @@
 import org.testcontainers.images.PullPolicy;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.MountableFile;
 
+import java.io.File;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.HttpURLConnection;
@@ -70,13 +72,13 @@ public abstract class AbstractAxonServerIntegrationTest {
 
     private static String getDockerImageName() {
         String envVariable = System.getenv("AXON_SERVER_IMAGE");
-        return envVariable != null ? envVariable : System.getProperty("AXON_SERVER_IMAGE", "axoniq/axonserver");
+        return envVariable != null ? envVariable : System.getProperty("AXON_SERVER_IMAGE", "docker.io/axoniq/axonserver");
     }
 
     @SuppressWarnings("resource")
     @Container
     public static GenericContainer<?> toxiProxyContainer =
-            new GenericContainer<>("shopify/toxiproxy")
+            new GenericContainer<>("docker.io/shopify/toxiproxy")
                     .withExposedPorts(8474, 8124)
                     .withImagePullPolicy(PullPolicy.ageBased(Duration.ofDays(1)))
                     .withNetwork(network)

src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java

@@ -463,7 +463,7 @@ void addTags() {
         assertEquals(2, updatedResponse.getTagCount());
         assertTrue(updatedResponse.getTagList().contains(initialTag));
         assertTrue(updatedResponse.getTagList().contains(newTag));
-    } 
+    }
 
     @Test
     void removeTags() {