From 98d1bfb29a3005fd44998190d930eabe4df2484b Mon Sep 17 00:00:00 2001 From: Silvano Biemans Date: Wed, 30 Jul 2025 10:31:19 +0200 Subject: [PATCH 1/7] Add a snapshot channel --- .../connector/AxonServerConnection.java | 3 + .../connector/event/SnapshotChannel.java | 20 + .../event/impl/SnapshotChannelImpl.java | 107 ++++ .../connector/impl/ContextConnection.java | 15 + .../AbstractAxonServerIntegrationTest.java | 7 +- .../connector/event/dcb/DcbEndToEndTest.java | 27 +- .../event/dcb/DcbSnapshotChannelTest.java | 584 ++++++++++++++++++ 7 files changed, 756 insertions(+), 7 deletions(-) create mode 100644 src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java create mode 100644 src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java create mode 100644 src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java diff --git a/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java b/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java index bd10b29a..d690c0d3 100644 --- a/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java +++ b/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,8 @@ public interface AxonServerConnection { */ QueryChannel queryChannel(); + SnapshotChannel snapshotChannel(); + /** * Returns the channel on which Event transformations can be performed with AxonServer. * diff --git a/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java new file mode 100644 index 00000000..e7ddca8e --- /dev/null +++ b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java @@ -0,0 +1,20 @@ +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; + +public interface SnapshotChannel { + CompletableFuture addSnapshot(AddSnapshotRequest request); + CompletableFuture deleteSnapshots(DeleteSnapshotsRequest request); + ResultStream listSnapshots(ListSnapshotsRequest request); + CompletableFuture getLastSnapshot(GetLastSnapshotRequest request); +} diff --git a/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java new file mode 100644 index 00000000..7de662db --- /dev/null +++ b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java @@ -0,0 +1,107 @@ +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.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; + +public class SnapshotChannelImpl extends AbstractAxonServerChannel implements SnapshotChannel { + + private final DcbSnapshotStoreGrpc.DcbSnapshotStoreStub snapshotStore; + private final ClientIdentification clientIdentification; + + 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 addSnapshot(AddSnapshotRequest request) { + FutureStreamObserver res = new FutureStreamObserver<>(null); + this.snapshotStore.add(request, res); + return res; + } + + @Override + public CompletableFuture deleteSnapshots(DeleteSnapshotsRequest request) { + FutureStreamObserver res = new FutureStreamObserver<>(null); + this.snapshotStore.delete(request, res); + return res; + } + + @Override + public ResultStream listSnapshots(ListSnapshotsRequest request) { + AbstractBufferedStream result = + new AbstractBufferedStream<>(clientIdentification.getClientId(), + BUFFER_SIZE, + REFILL_BATCH) { + @Override + protected ListSnapshotsResponse terminalMessage() { + return ListSnapshotsResponse.newBuilder() + .build(); + } + + @Override + protected Empty buildFlowControlMessage(FlowControl flowControl) { + return null; + } + }; + snapshotStore.list(request, result); + return result; + } + + @Override + public CompletableFuture getLastSnapshot(GetLastSnapshotRequest request) { + FutureStreamObserver res = new FutureStreamObserver<>(null); + this.snapshotStore.getLast(request, res); + return res; + } + + @Override + public void connect() { + + } + + @Override + public void reconnect() { + + } + + @Override + public void disconnect() { + + } + + @Override + public boolean isReady() { + return false; + } +} diff --git a/src/main/java/io/axoniq/axonserver/connector/impl/ContextConnection.java b/src/main/java/io/axoniq/axonserver/connector/impl/ContextConnection.java index c6454696..75c3c22b 100644 --- a/src/main/java/io/axoniq/axonserver/connector/impl/ContextConnection.java +++ b/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 queryChannel = new AtomicReference<>(); private final AtomicReference eventTransformationChannel = new AtomicReference<>(); private final AtomicReference adminChannel = new AtomicReference<>(); + private final AtomicReference 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, diff --git a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java index 9a2802f2..c88b1b59 100644 --- a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java +++ b/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; @@ -63,6 +65,7 @@ public abstract class AbstractAxonServerIntegrationTest { .withEnv("AXONIQ_AXONSERVER_DEVMODE_ENABLED", "true") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_TOKEN", "user-token") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_ADMIN_TOKEN", "admin-token") + .withCopyFileToContainer(MountableFile.forHostPath("/home/silvano/IdeaProjects/axon-server/package-axonserver-enterprise/target/axonserver.jar"), "/axonserver/axonserver.jar") .withImagePullPolicy(PullPolicy.ageBased(Duration.ofDays(1))) .withNetwork(network) .withNetworkAliases("axonserver") @@ -70,13 +73,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) diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java index adad4da1..cf2f46b8 100644 --- a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java @@ -7,19 +7,19 @@ import io.axoniq.axonserver.connector.AxonServerConnectionFactory; import io.axoniq.axonserver.connector.ResultStream; import io.axoniq.axonserver.connector.ResultStreamPublisher; +import io.axoniq.axonserver.connector.event.SnapshotChannel; import io.axoniq.axonserver.connector.impl.ServerAddress; import io.axoniq.axonserver.connector.event.DcbEventChannel; +import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotRequest; import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition; import io.axoniq.axonserver.grpc.event.dcb.Criterion; import io.axoniq.axonserver.grpc.event.dcb.Event; -import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest; -import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest; +import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotRequest; import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse; -import io.axoniq.axonserver.grpc.event.dcb.GetTagsRequest; import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse; -import io.axoniq.axonserver.grpc.event.dcb.GetTailRequest; import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse; +import io.axoniq.axonserver.grpc.event.dcb.Snapshot; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest; @@ -34,7 +34,6 @@ import reactor.core.publisher.Flux; import reactor.test.StepVerifier; -import java.io.IOException; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; @@ -142,6 +141,24 @@ static void initializeLocal() { } } + @Test + void addASnapshot() throws InterruptedException { + SnapshotChannel snapshotChannel = connection.snapshotChannel(); + + AddSnapshotRequest addSnapshotRequest = AddSnapshotRequest.newBuilder() + .setKey(ByteString.copyFrom("Hello".getBytes())) + .setSnapshot(Snapshot.newBuilder() + .setName("A key") + .setRevision("1.0") + .setPayload(ByteString.copyFrom("A payload".getBytes())).build()) + .setPrune(false) + .setSequence(0) + .build(); + snapshotChannel.addSnapshot(addSnapshotRequest).join(); + + System.out.println(snapshotChannel.getLastSnapshot(GetLastSnapshotRequest.newBuilder().setKey(ByteString.copyFrom("Hello".getBytes())).build()).join().getSnapshot().getName()); + } + @Test void streamWhileAppendingWithASingleTagCriterion() { long head = retrieveHead(); diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java new file mode 100644 index 00000000..a3a50c4f --- /dev/null +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java @@ -0,0 +1,584 @@ +package io.axoniq.axonserver.connector.event.dcb; + +import com.google.protobuf.ByteString; +import io.axoniq.axonserver.connector.AbstractAxonServerIntegrationTest; +import io.axoniq.axonserver.connector.AxonServerConnection; +import io.axoniq.axonserver.connector.AxonServerConnectionFactory; +import io.axoniq.axonserver.connector.ResultStreamPublisher; +import io.axoniq.axonserver.connector.event.SnapshotChannel; +import io.axoniq.axonserver.connector.impl.ServerAddress; +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.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 io.axoniq.axonserver.grpc.event.dcb.Snapshot; +import org.junit.jupiter.api.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import java.util.stream.LongStream; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.junit.jupiter.api.Assertions.*; + +class DcbSnapshotChannelTest extends AbstractAxonServerIntegrationTest { + + private static final Logger logger = LoggerFactory.getLogger(DcbSnapshotChannelTest.class); + private static final boolean LOCAL = false; + + private AxonServerConnection connection; + private AxonServerConnectionFactory client; + private SnapshotChannel snapshotChannel; + + @BeforeEach + void setUp() { + AxonServerConnectionFactory.Builder builder = AxonServerConnectionFactory.forClient("dcb-snapshot-test") + .connectTimeout(1500, + TimeUnit.MILLISECONDS) + .reconnectInterval(500, MILLISECONDS); + + if (LOCAL) { + builder.routingServers(new ServerAddress("localhost", 8124)); + logger.info("Using local Axon Server instance at localhost:8124"); + } else { + builder.routingServers(axonServerAddress); + } + + client = builder.build(); + connection = client.connect("default"); + snapshotChannel = connection.snapshotChannel(); + } + + @AfterEach + void tearDown() { + client.shutdown(); + } + + @Override + protected boolean dcbContext() { + return true; + } + + // Basic Put/Overwrite Semantics Tests + + @Test + void addSnapshotWithSameSequenceReplacesExisting() { + ByteString key = ByteString.copyFrom("replace-test".getBytes()); + ByteString value1 = ByteString.copyFrom("first".getBytes()); + ByteString value2 = ByteString.copyFrom("second".getBytes()); + long sequence = 1L; + + // Add first snapshot + AddSnapshotRequest request1 = AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(sequence) + .setPrune(false) + .setSnapshot(Snapshot.newBuilder() + .setName("snapshot1") + .setRevision("1.0") + .setPayload(value1) + .build()) + .build(); + + snapshotChannel.addSnapshot(request1).join(); + + // Add second snapshot with same sequence + AddSnapshotRequest request2 = AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(sequence) + .setPrune(false) + .setSnapshot(Snapshot.newBuilder() + .setName("snapshot2") + .setRevision("1.0") + .setPayload(value2) + .build()) + .build(); + + snapshotChannel.addSnapshot(request2).join(); + + // Verify only second snapshot exists + GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + assertNotNull(last); + assertEquals(sequence, last.getSequence()); + assertEquals(value2, last.getSnapshot().getPayload()); + + // Verify through list + List snapshots = collectSnapshots(key, 0, 2); + assertEquals(1, snapshots.size()); + assertEquals(value2, snapshots.get(0).getSnapshot().getPayload()); + } + + @Test + void addNewSnapshotToEmptyStoreIsRetrievable() { + ByteString key = ByteString.copyFrom("new-snapshot".getBytes()); + ByteString value = ByteString.copyFrom("test-value".getBytes()); + long sequence = 42L; + + AddSnapshotRequest request = AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(sequence) + .setPrune(false) + .setSnapshot(Snapshot.newBuilder() + .setName("test-snapshot") + .setRevision("1.0") + .setPayload(value) + .build()) + .build(); + + snapshotChannel.addSnapshot(request).join(); + + GetLastSnapshotResponse result = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + assertNotNull(result); + assertEquals(sequence, result.getSequence()); + assertEquals(value, result.getSnapshot().getPayload()); + } + + // Pruning Behavior Tests + + @Test + void addingSnapshotWithPruneEnabledKeepsOnlySnapshotsGreaterOrEqualToAddedSequence() { + ByteString key = ByteString.copyFrom("prune-test".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add snapshots at sequences 50, 100, 150 + addSnapshot(key, 50L, false, value); + addSnapshot(key, 100L, false, value); + addSnapshot(key, 150L, false, value); + System.out.println("Got here"); + // Add at sequence 75 with pruning - should keep 75, 100, 150 + addSnapshot(key, 75L, true, value); + + List snapshots = collectSnapshots(key, 0, 200); + List sequences = snapshots.stream() + .map(ListSnapshotsResponse::getSequence) + .sorted() + .collect(Collectors.toList()); + + assertEquals(List.of(75L, 100L, 150L), sequences); + } + + @Test + void pruningOnlyAffectsSnapshotsWithSameKey() { + ByteString key1 = ByteString.copyFrom("key1".getBytes()); + ByteString key2 = ByteString.copyFrom("key2".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add snapshots to both keys + addSnapshot(key1, 50L, false, value); + addSnapshot(key1, 100L, false, value); + addSnapshot(key2, 25L, false, value); + addSnapshot(key2, 75L, false, value); + + // Prune key1 at sequence 80 + addSnapshot(key1, 80L, true, value); + + // Check key1 - should have 80 and 100 + List snapshots1 = collectSnapshots(key1, 0, 200); + List sequences1 = snapshots1.stream() + .map(ListSnapshotsResponse::getSequence) + .sorted() + .collect(Collectors.toList()); + assertEquals(List.of(80L, 100L), sequences1); + + // Check key2 - should be unchanged + List snapshots2 = collectSnapshots(key2, 0, 200); + List sequences2 = snapshots2.stream() + .map(ListSnapshotsResponse::getSequence) + .sorted() + .collect(Collectors.toList()); + assertEquals(List.of(25L, 75L), sequences2); + } + + // Non-Pruning Behavior Tests + + @Test + void addingSubsequentSnapshotsWithPruneDisabledKeepsAllSnapshots() throws InterruptedException { + ByteString key = ByteString.copyFrom("no-prune".getBytes()); + ByteString value = ByteString.copyFrom("world".getBytes()); + int limit = 3; + CountDownLatch latch = new CountDownLatch(limit); + + for (long i = 0; i < limit; i++) { + long seq = i; + CompletableFuture future = snapshotChannel.addSnapshot( + AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(seq) + .setPrune(false) + .setSnapshot(Snapshot.newBuilder() + .setName("snapshot-" + seq) + .setRevision("1.0") + .setPayload(value) + .build()) + .build() + ); + future.thenRun(latch::countDown); + } + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + + GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + assertNotNull(last); + assertEquals(limit - 1, last.getSequence()); + + List snapshots = collectSnapshots(key, 0, limit); + assertEquals(limit, snapshots.size()); + } + + // Delete Operation Tests + + @Test + void deleteRangeRemovesSnapshotsWithinInclusiveExclusiveBounds() { + ByteString key = ByteString.copyFrom("delete-range".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add snapshots at 50, 100, 150, 200 + List sequences = List.of(50L, 100L, 150L, 200L); + sequences.forEach(seq -> addSnapshot(key, seq, false, value)); + + // Delete from 100 to 200 (should delete 100 and 150, not 200) + snapshotChannel.deleteSnapshots( + DeleteSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(100) + .setToSequence(200) + .build() + ).join(); + + List snapshots = collectSnapshots(key, 0, 250); + List remainingSequences = snapshots.stream() + .map(ListSnapshotsResponse::getSequence) + .sorted() + .collect(Collectors.toList()); + + assertEquals(List.of(50L, 200L), remainingSequences); + } + + @Test + void deletingAllSnapshotsMakesGetLastReturnNull() { + ByteString key = ByteString.copyFrom("delete-all".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add some snapshots + addSnapshot(key, 50L, false, value); + addSnapshot(key, 100L, false, value); + + // Delete all + snapshotChannel.deleteSnapshots( + DeleteSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(0) + .setToSequence(1000) + .build() + ).join(); + + // GetLast should return empty response + GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + // The response should have no snapshot set + assertFalse(last.hasSnapshot()); + } + + @Test + void sequenceBecomesImmediatelyReusableAfterDeletion() { + ByteString key = ByteString.copyFrom("reuse-sequence".getBytes()); + ByteString value1 = ByteString.copyFrom("original".getBytes()); + ByteString value2 = ByteString.copyFrom("reused".getBytes()); + long sequence = 100L; + + // Add snapshot + addSnapshot(key, sequence, false, value1); + + // Verify it exists + GetLastSnapshotResponse first = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertEquals(value1, first.getSnapshot().getPayload()); + + // Delete it + snapshotChannel.deleteSnapshots( + DeleteSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(sequence) + .setToSequence(sequence + 1) + .build() + ).join(); + + // Verify it's gone + GetLastSnapshotResponse afterDelete = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertFalse(afterDelete.hasSnapshot()); + + // Reuse the same sequence + addSnapshot(key, sequence, false, value2); + + // Verify new snapshot is there + GetLastSnapshotResponse reused = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertTrue(reused.hasSnapshot()); + assertEquals(sequence, reused.getSequence()); + assertEquals(value2, reused.getSnapshot().getPayload()); + } + + // List Operation Tests + + @Test + void listReturnsEmptyWhenNoSnapshotsExistInRange() { + ByteString key = ByteString.copyFrom("empty-range".getBytes()); + + List snapshots = collectSnapshots(key, 100, 200); + assertEquals(0, snapshots.size()); + } + + @Test + void listReturnsSnapshotsOrderedBySequence() { + ByteString key = ByteString.copyFrom("order-test".getBytes()); + ByteString value = ByteString.copyFrom("val".getBytes()); + + // Add in random order + List sequences = List.of(150L, 50L, 200L, 100L, 75L); + sequences.forEach(seq -> addSnapshot(key, seq, false, value)); + + List snapshots = collectSnapshots(key, 0, 300); + List returnedSequences = snapshots.stream() + .map(ListSnapshotsResponse::getSequence) + .collect(Collectors.toList()); + + assertEquals(returnedSequences.stream().sorted().collect(Collectors.toList()), returnedSequences); + } + + @Test + void listRespectsInclusiveLowerAndExclusiveUpperBounds() { + ByteString key = ByteString.copyFrom("bounds-test".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add snapshots at boundaries and beyond + List sequences = List.of(25L, 50L, 75L, 100L, 125L); + sequences.forEach(seq -> addSnapshot(key, seq, false, value)); + + List snapshots = collectSnapshots(key, 50, 100); + List returnedSequences = snapshots.stream() + .map(ListSnapshotsResponse::getSequence) + .sorted() + .collect(Collectors.toList()); + + assertEquals(List.of(50L, 75L), returnedSequences); + } + + // GetLast Operation Tests + + @Test + void getLastReturnsNullWhenNoSnapshotExistsForKey() { + ByteString key = ByteString.copyFrom("unknown-key".getBytes()); + + GetLastSnapshotResponse result = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + assertFalse(result.hasSnapshot()); + } + + @Test + void getLastReturnsHighestSequenceAfterMixedAddAndDeleteOperations() { + ByteString key = ByteString.copyFrom("getlast-mixed".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Add snapshots + addSnapshot(key, 50L, false, value); + addSnapshot(key, 150L, false, value); + addSnapshot(key, 100L, false, value); + addSnapshot(key, 200L, false, value); + + // Delete highest + snapshotChannel.deleteSnapshots( + DeleteSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(200) + .setToSequence(201) + .build() + ).join(); + + GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + + assertTrue(last.hasSnapshot()); + assertEquals(150L, last.getSequence()); + } + + // Complex Scenarios + + @Test + void switchingFromNonPruningToPruningStrategyCleansUpOldSnapshots() { + ByteString key = ByteString.copyFrom("strategy-switch".getBytes()); + ByteString value = ByteString.copyFrom("data".getBytes()); + + // Accumulate many snapshots without pruning + LongStream.range(0, 100).forEach(seq -> addSnapshot(key, seq, false, value)); + + // Verify all exist + List beforeSnapshots = collectSnapshots(key, 0, 100); + assertEquals(100, beforeSnapshots.size()); + + // Switch to pruning strategy + addSnapshot(key, 95L, true, value); + + // Verify cleanup + List afterSnapshots = collectSnapshots(key, 0, 100); + assertEquals(5, afterSnapshots.size()); // 95, 96, 97, 98, 99 + + long minSequence = afterSnapshots.stream() + .mapToLong(ListSnapshotsResponse::getSequence) + .min() + .orElse(-1); + assertEquals(95L, minSequence); + } + + @Test + void addDeleteAddSameSequenceDemonstratesSequenceReusability() { + ByteString key = ByteString.copyFrom("add-delete-add".getBytes()); + ByteString value1 = ByteString.copyFrom("first".getBytes()); + ByteString value2 = ByteString.copyFrom("second".getBytes()); + long sequence = 50L; + + // Add + addSnapshot(key, sequence, false, value1); + + // Verify + GetLastSnapshotResponse first = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertEquals(value1, first.getSnapshot().getPayload()); + + // Delete + snapshotChannel.deleteSnapshots( + DeleteSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(sequence) + .setToSequence(sequence + 1) + .build() + ).join(); + + // Verify deletion + GetLastSnapshotResponse deleted = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertFalse(deleted.hasSnapshot()); + + // Add again with same sequence + addSnapshot(key, sequence, false, value2); + + // Verify new value + GetLastSnapshotResponse second = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertTrue(second.hasSnapshot()); + assertEquals(sequence, second.getSequence()); + assertEquals(value2, second.getSnapshot().getPayload()); + } + + // Performance Tests (disabled by default) + + @Test +// @Disabled("Performance test - enable manually") + void appendingTwoMillionSnapshotsWithPruneEnabledCompletesInReasonableTime() throws InterruptedException { + ByteString key = ByteString.copyFrom("perf-prune".getBytes()); + ByteString value = ByteString.copyFrom("payload".getBytes()); + long totalSnapshots = 2_000_000L; + + long startTime = System.currentTimeMillis(); + + List> futures = new ArrayList<>(); + for (long i = 0; i < totalSnapshots; i++) { + if (i > 0 && i % 1_000_000 == 0) { + logger.info("Written {} snapshots...", i); + } + futures.add(snapshotChannel.addSnapshot( + AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(i) + .setPrune(true) + .setSnapshot(Snapshot.newBuilder() + .setName("snapshot") + .setRevision("1.0") + .setPayload(value) + .build()) + .build() + )); + } + + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + long duration = (System.currentTimeMillis() - startTime) / 1000; + logger.info("Completed 2M writes with prune in {} seconds", duration); + + // Verify only last snapshot remains + GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + assertTrue(last.hasSnapshot()); + assertEquals(totalSnapshots - 1, last.getSequence()); + + List snapshots = collectSnapshots(key, 0, totalSnapshots); + assertEquals(1, snapshots.size()); + } + + // Helper methods + + private void addSnapshot(ByteString key, long sequence, boolean prune, ByteString value) { + snapshotChannel.addSnapshot( + AddSnapshotRequest.newBuilder() + .setKey(key) + .setSequence(sequence) + .setPrune(prune) + .setSnapshot(Snapshot.newBuilder() + .setName("snapshot-" + sequence) + .setRevision("1.0") + .setPayload(value) + .build()) + .build() + ).join(); + } + + private List collectSnapshots(ByteString key, long fromSequence, long toSequence) { + ListSnapshotsRequest request = ListSnapshotsRequest.newBuilder() + .setKey(key) + .setFromSequence(fromSequence) + .setToSequence(toSequence) + .build(); + + AtomicReference> result = new AtomicReference<>(new ArrayList<>()); + + StepVerifier.create(Flux.from(new ResultStreamPublisher<>(() -> snapshotChannel.listSnapshots(request)))) + .recordWith(() -> result.get()) + .thenConsumeWhile(response -> response.hasSnapshot()) + .verifyComplete(); + + return result.get(); + } +} \ No newline at end of file From c7d2d22d78e9ef6ea43140944e5ba82d4d1fd362 Mon Sep 17 00:00:00 2001 From: Milan Savic Date: Mon, 4 Aug 2025 13:54:30 +0200 Subject: [PATCH 2/7] - Added JavaDoc to the `SnapshotChannel` --- .../connector/event/SnapshotChannel.java | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java index e7ddca8e..f826e190 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java @@ -12,9 +12,49 @@ import java.util.concurrent.CompletableFuture; +/** + * Communication channel for Snapshot-related interactions with AxonServer. This interface provides operations for + * managing snapshots in a Dynamic Consistency Boundaries (DCB) environment, 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 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 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 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 getLastSnapshot(GetLastSnapshotRequest request); } From 7c793f2212cf83f31a55fedcecb4c037d84dc711 Mon Sep 17 00:00:00 2001 From: Silvano Biemans Date: Wed, 27 Aug 2025 11:12:38 +0200 Subject: [PATCH 3/7] Set isready in DcbSnapshotChannelImpl --- .../axonserver/connector/event/impl/SnapshotChannelImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java index 7de662db..25fb9e25 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java @@ -102,6 +102,6 @@ public void disconnect() { @Override public boolean isReady() { - return false; + return true; } } From c78a4137b9490b7ea3bd4192111b07218a7990a2 Mon Sep 17 00:00:00 2001 From: Marco Amann Date: Wed, 10 Sep 2025 15:40:03 +0200 Subject: [PATCH 4/7] Adjust tests for changes in api --- pom.xml | 2 +- .../AbstractAxonServerIntegrationTest.java | 2 +- .../connector/event/dcb/DcbEndToEndTest.java | 2 +- .../event/dcb/DcbSnapshotChannelTest.java | 94 +++++++------------ 4 files changed, 37 insertions(+), 63 deletions(-) diff --git a/pom.xml b/pom.xml index e0ce5c56..7fb4fcdd 100644 --- a/pom.xml +++ b/pom.xml @@ -44,7 +44,7 @@ axoniq https://sonarcloud.io - 2025.1.1 + 2025.1.2-SNAPSHOT 1.69.0 4.31.1 diff --git a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java index c88b1b59..5e5bc8bb 100644 --- a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java @@ -65,7 +65,7 @@ public abstract class AbstractAxonServerIntegrationTest { .withEnv("AXONIQ_AXONSERVER_DEVMODE_ENABLED", "true") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_TOKEN", "user-token") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_ADMIN_TOKEN", "admin-token") - .withCopyFileToContainer(MountableFile.forHostPath("/home/silvano/IdeaProjects/axon-server/package-axonserver-enterprise/target/axonserver.jar"), "/axonserver/axonserver.jar") + .withCopyFileToContainer(MountableFile.forHostPath("/home/marco/Documents/AxonIQ/AxonServer/axon-server/package-axonserver-enterprise/target/axonserver.jar"), "/axonserver/axonserver.jar") .withImagePullPolicy(PullPolicy.ageBased(Duration.ofDays(1))) .withNetwork(network) .withNetworkAliases("axonserver") diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java index 8f2f0a57..dac6cf7b 100644 --- a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java @@ -149,7 +149,7 @@ void addASnapshot() throws InterruptedException { .setKey(ByteString.copyFrom("Hello".getBytes())) .setSnapshot(Snapshot.newBuilder() .setName("A key") - .setRevision("1.0") + .setVersion("1.0") .setPayload(ByteString.copyFrom("A payload".getBytes())).build()) .setPrune(false) .setSequence(0) diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java index a3a50c4f..cfdbbc28 100644 --- a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java @@ -15,6 +15,7 @@ import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsRequest; import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsResponse; import io.axoniq.axonserver.grpc.event.dcb.Snapshot; +import io.grpc.StatusRuntimeException; import org.junit.jupiter.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,8 +23,10 @@ import reactor.test.StepVerifier; import java.util.ArrayList; +import java.util.Comparator; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -87,7 +90,7 @@ void addSnapshotWithSameSequenceReplacesExisting() { .setPrune(false) .setSnapshot(Snapshot.newBuilder() .setName("snapshot1") - .setRevision("1.0") + .setVersion("1.0") .setPayload(value1) .build()) .build(); @@ -101,7 +104,7 @@ void addSnapshotWithSameSequenceReplacesExisting() { .setPrune(false) .setSnapshot(Snapshot.newBuilder() .setName("snapshot2") - .setRevision("1.0") + .setVersion("1.0") .setPayload(value2) .build()) .build(); @@ -135,7 +138,8 @@ void addNewSnapshotToEmptyStoreIsRetrievable() { .setPrune(false) .setSnapshot(Snapshot.newBuilder() .setName("test-snapshot") - .setRevision("1.0") + .setVersion("1.0") + .setTimestamp(System.currentTimeMillis()) .setPayload(value) .build()) .build(); @@ -225,7 +229,7 @@ void addingSubsequentSnapshotsWithPruneDisabledKeepsAllSnapshots() throws Interr .setPrune(false) .setSnapshot(Snapshot.newBuilder() .setName("snapshot-" + seq) - .setRevision("1.0") + .setVersion("1.0") .setPayload(value) .build()) .build() @@ -249,7 +253,7 @@ void addingSubsequentSnapshotsWithPruneDisabledKeepsAllSnapshots() throws Interr // Delete Operation Tests @Test - void deleteRangeRemovesSnapshotsWithinInclusiveExclusiveBounds() { + void deleteRangeRemovesSnapshotsWithinExclusiveBounds() { ByteString key = ByteString.copyFrom("delete-range".getBytes()); ByteString value = ByteString.copyFrom("data".getBytes()); @@ -261,7 +265,6 @@ void deleteRangeRemovesSnapshotsWithinInclusiveExclusiveBounds() { snapshotChannel.deleteSnapshots( DeleteSnapshotsRequest.newBuilder() .setKey(key) - .setFromSequence(100) .setToSequence(200) .build() ).join(); @@ -272,11 +275,11 @@ void deleteRangeRemovesSnapshotsWithinInclusiveExclusiveBounds() { .sorted() .collect(Collectors.toList()); - assertEquals(List.of(50L, 200L), remainingSequences); + assertEquals(List.of(200L), remainingSequences); } @Test - void deletingAllSnapshotsMakesGetLastReturnNull() { + void deletingAllSnapshotsMakesGetLastThrowException() { ByteString key = ByteString.copyFrom("delete-all".getBytes()); ByteString value = ByteString.copyFrom("data".getBytes()); @@ -288,18 +291,17 @@ void deletingAllSnapshotsMakesGetLastReturnNull() { snapshotChannel.deleteSnapshots( DeleteSnapshotsRequest.newBuilder() .setKey(key) - .setFromSequence(0) .setToSequence(1000) .build() ).join(); // GetLast should return empty response - GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( - GetLastSnapshotRequest.newBuilder().setKey(key).build() - ).join(); + assertThrows(CompletionException.class, ()->{ + snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + }); - // The response should have no snapshot set - assertFalse(last.hasSnapshot()); } @Test @@ -322,16 +324,16 @@ void sequenceBecomesImmediatelyReusableAfterDeletion() { snapshotChannel.deleteSnapshots( DeleteSnapshotsRequest.newBuilder() .setKey(key) - .setFromSequence(sequence) .setToSequence(sequence + 1) .build() ).join(); // Verify it's gone - GetLastSnapshotResponse afterDelete = snapshotChannel.getLastSnapshot( - GetLastSnapshotRequest.newBuilder().setKey(key).build() - ).join(); - assertFalse(afterDelete.hasSnapshot()); + assertThrows(CompletionException.class, ()->{ + snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + }); // Reuse the same sequence addSnapshot(key, sequence, false, value2); @@ -369,7 +371,7 @@ void listReturnsSnapshotsOrderedBySequence() { .map(ListSnapshotsResponse::getSequence) .collect(Collectors.toList()); - assertEquals(returnedSequences.stream().sorted().collect(Collectors.toList()), returnedSequences); + assertEquals(returnedSequences.stream().sorted(Comparator.reverseOrder()).collect(Collectors.toList()), returnedSequences); } @Test @@ -396,39 +398,11 @@ void listRespectsInclusiveLowerAndExclusiveUpperBounds() { void getLastReturnsNullWhenNoSnapshotExistsForKey() { ByteString key = ByteString.copyFrom("unknown-key".getBytes()); - GetLastSnapshotResponse result = snapshotChannel.getLastSnapshot( - GetLastSnapshotRequest.newBuilder().setKey(key).build() - ).join(); - - assertFalse(result.hasSnapshot()); - } - - @Test - void getLastReturnsHighestSequenceAfterMixedAddAndDeleteOperations() { - ByteString key = ByteString.copyFrom("getlast-mixed".getBytes()); - ByteString value = ByteString.copyFrom("data".getBytes()); - - // Add snapshots - addSnapshot(key, 50L, false, value); - addSnapshot(key, 150L, false, value); - addSnapshot(key, 100L, false, value); - addSnapshot(key, 200L, false, value); - - // Delete highest - snapshotChannel.deleteSnapshots( - DeleteSnapshotsRequest.newBuilder() - .setKey(key) - .setFromSequence(200) - .setToSequence(201) - .build() - ).join(); - - GetLastSnapshotResponse last = snapshotChannel.getLastSnapshot( - GetLastSnapshotRequest.newBuilder().setKey(key).build() - ).join(); - - assertTrue(last.hasSnapshot()); - assertEquals(150L, last.getSequence()); + assertThrows(CompletionException.class, ()->{ + snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + }); } // Complex Scenarios @@ -479,16 +453,16 @@ void addDeleteAddSameSequenceDemonstratesSequenceReusability() { snapshotChannel.deleteSnapshots( DeleteSnapshotsRequest.newBuilder() .setKey(key) - .setFromSequence(sequence) .setToSequence(sequence + 1) .build() ).join(); // Verify deletion - GetLastSnapshotResponse deleted = snapshotChannel.getLastSnapshot( - GetLastSnapshotRequest.newBuilder().setKey(key).build() - ).join(); - assertFalse(deleted.hasSnapshot()); + assertThrows(CompletionException.class, ()->{ + snapshotChannel.getLastSnapshot( + GetLastSnapshotRequest.newBuilder().setKey(key).build() + ).join(); + }); // Add again with same sequence addSnapshot(key, sequence, false, value2); @@ -525,7 +499,7 @@ void appendingTwoMillionSnapshotsWithPruneEnabledCompletesInReasonableTime() thr .setPrune(true) .setSnapshot(Snapshot.newBuilder() .setName("snapshot") - .setRevision("1.0") + .setVersion("1.0") .setPayload(value) .build()) .build() @@ -558,7 +532,7 @@ private void addSnapshot(ByteString key, long sequence, boolean prune, ByteStrin .setPrune(prune) .setSnapshot(Snapshot.newBuilder() .setName("snapshot-" + sequence) - .setRevision("1.0") + .setVersion("1.0") .setPayload(value) .build()) .build() From 5f77593e941c2e795fc4863a1272fef52f99e532 Mon Sep 17 00:00:00 2001 From: Silvano Biemans Date: Tue, 23 Sep 2025 08:21:32 +0200 Subject: [PATCH 5/7] Remove copying local axon server jar to docker container --- .../connector/AbstractAxonServerIntegrationTest.java | 1 - .../axonserver/connector/event/dcb/DcbSnapshotChannelTest.java | 3 +-- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java index 5e5bc8bb..a9368b9b 100644 --- a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java @@ -65,7 +65,6 @@ public abstract class AbstractAxonServerIntegrationTest { .withEnv("AXONIQ_AXONSERVER_DEVMODE_ENABLED", "true") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_TOKEN", "user-token") .withEnv("AXONIQ_AXONSERVER_ACCESSCONTROL_ADMIN_TOKEN", "admin-token") - .withCopyFileToContainer(MountableFile.forHostPath("/home/marco/Documents/AxonIQ/AxonServer/axon-server/package-axonserver-enterprise/target/axonserver.jar"), "/axonserver/axonserver.jar") .withImagePullPolicy(PullPolicy.ageBased(Duration.ofDays(1))) .withNetwork(network) .withNetworkAliases("axonserver") diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java index cfdbbc28..4323796a 100644 --- a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbSnapshotChannelTest.java @@ -15,7 +15,6 @@ import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsRequest; import io.axoniq.axonserver.grpc.event.dcb.ListSnapshotsResponse; import io.axoniq.axonserver.grpc.event.dcb.Snapshot; -import io.grpc.StatusRuntimeException; import org.junit.jupiter.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -479,7 +478,7 @@ void addDeleteAddSameSequenceDemonstratesSequenceReusability() { // Performance Tests (disabled by default) @Test -// @Disabled("Performance test - enable manually") + @Disabled("Performance test - enable manually") void appendingTwoMillionSnapshotsWithPruneEnabledCompletesInReasonableTime() throws InterruptedException { ByteString key = ByteString.copyFrom("perf-prune".getBytes()); ByteString value = ByteString.copyFrom("payload".getBytes()); From b966d9a56d56680d24c3e1ee7d69f0a4c8df206e Mon Sep 17 00:00:00 2001 From: Milan Savic Date: Fri, 24 Oct 2025 13:22:13 +0200 Subject: [PATCH 6/7] Removed unnecessary test from DcbEndToEndTest. --- .../connector/event/dcb/DcbEndToEndTest.java | 22 ------------------- 1 file changed, 22 deletions(-) diff --git a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java index 5c246f8e..8de91356 100644 --- a/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java @@ -8,18 +8,14 @@ import io.axoniq.axonserver.connector.ResultStream; import io.axoniq.axonserver.connector.ResultStreamPublisher; import io.axoniq.axonserver.connector.event.DcbEventChannel; -import io.axoniq.axonserver.connector.event.SnapshotChannel; import io.axoniq.axonserver.connector.impl.ServerAddress; -import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotRequest; import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition; import io.axoniq.axonserver.grpc.event.dcb.Criterion; import io.axoniq.axonserver.grpc.event.dcb.Event; -import io.axoniq.axonserver.grpc.event.dcb.GetLastSnapshotRequest; import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse; import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse; import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse; -import io.axoniq.axonserver.grpc.event.dcb.Snapshot; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest; @@ -136,24 +132,6 @@ static void initializeLocal() { } } - @Test - void addASnapshot() throws InterruptedException { - SnapshotChannel snapshotChannel = connection.snapshotChannel(); - - AddSnapshotRequest addSnapshotRequest = AddSnapshotRequest.newBuilder() - .setKey(ByteString.copyFrom("Hello".getBytes())) - .setSnapshot(Snapshot.newBuilder() - .setName("A key") - .setVersion("1.0") - .setPayload(ByteString.copyFrom("A payload".getBytes())).build()) - .setPrune(false) - .setSequence(0) - .build(); - snapshotChannel.addSnapshot(addSnapshotRequest).join(); - - System.out.println(snapshotChannel.getLastSnapshot(GetLastSnapshotRequest.newBuilder().setKey(ByteString.copyFrom("Hello".getBytes())).build()).join().getSnapshot().getName()); - } - @Test void streamWhileAppendingWithASingleTagCriterion() { long head = retrieveHead(); From cb105c7eb47252a32b68d7e63a49fb004cd5b04a Mon Sep 17 00:00:00 2001 From: Milan Savic Date: Fri, 24 Oct 2025 14:15:47 +0200 Subject: [PATCH 7/7] Added missing JavaDoc. Made SnapshotChannelImpl aware of inflight listings, and close them. --- .../connector/AxonServerConnection.java | 5 +++ .../connector/event/SnapshotChannel.java | 3 +- .../event/impl/SnapshotChannelImpl.java | 38 +++++++++++++++---- 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java b/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java index d690c0d3..5a7d8ab2 100644 --- a/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java +++ b/src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java @@ -97,6 +97,11 @@ public interface AxonServerConnection { */ QueryChannel queryChannel(); + /** + * Returns the channel on which Snapshotting can be performed with AxonServer. + * + * @return the channel for Snapshotting + */ SnapshotChannel snapshotChannel(); /** diff --git a/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java index f826e190..ba6bb78b 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/SnapshotChannel.java @@ -14,8 +14,7 @@ /** * Communication channel for Snapshot-related interactions with AxonServer. This interface provides operations for - * managing snapshots in a Dynamic Consistency Boundaries (DCB) environment, including adding, deleting, listing, and - * retrieving snapshots. + * managing snapshots, including adding, deleting, listing, and retrieving snapshots. */ public interface SnapshotChannel { diff --git a/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java index 25fb9e25..bb8c2203 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/impl/SnapshotChannelImpl.java @@ -19,20 +19,32 @@ 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 implements SnapshotChannel { private final DcbSnapshotStoreGrpc.DcbSnapshotStoreStub snapshotStore; private final ClientIdentification clientIdentification; + private final Set> 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 clientIdentification identifies a client * @param executor a {@link ScheduledExecutorService} used to schedule reconnections * @param axonServerManagedChannel the {@link AxonServerManagedChannel} used to connect to AxonServer */ @@ -59,7 +71,7 @@ public CompletableFuture deleteSnapshots(DeleteSnapshot @Override public ResultStream listSnapshots(ListSnapshotsRequest request) { - AbstractBufferedStream result = + AbstractBufferedStream buffer = new AbstractBufferedStream<>(clientIdentification.getClientId(), BUFFER_SIZE, REFILL_BATCH) { @@ -74,8 +86,15 @@ protected Empty buildFlowControlMessage(FlowControl flowControl) { return null; } }; - snapshotStore.list(request, result); - return result; + buffers.add(buffer); + buffer.onCloseRequested(() -> buffers.remove(buffer)); + try { + snapshotStore.list(request, buffer); + } catch (Exception e) { + buffers.remove(buffer); + throw e; + } + return buffer; } @Override @@ -87,21 +106,26 @@ public CompletableFuture getLastSnapshot(GetLastSnapsho @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(); + } }