From abbf8882ba82eaec2d9f0c9b6992bf53781ed694 Mon Sep 17 00:00:00 2001 From: Marc Date: Tue, 26 Aug 2025 14:24:36 +0200 Subject: [PATCH 1/3] add DCB event scheduler --- pom.xml | 2 +- .../connector/event/DcbEventChannel.java | 60 ++++++++++++++++++ .../event/impl/DcbEventChannelImpl.java | 57 +++++++++++++++++ .../connector/event/dcb/DcbEndToEndTest.java | 61 ++++++++++++++++--- 4 files changed, 170 insertions(+), 10 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/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java b/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java index a73433bc..f48346d5 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java @@ -17,9 +17,11 @@ package io.axoniq.axonserver.connector.event; import io.axoniq.axonserver.connector.ResultStream; +import io.axoniq.axonserver.grpc.InstructionAck; import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse; import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition; +import io.axoniq.axonserver.grpc.event.dcb.Event; import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse; import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse; import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse; @@ -32,6 +34,7 @@ import io.axoniq.axonserver.grpc.event.dcb.Tag; import io.axoniq.axonserver.grpc.event.dcb.TaggedEvent; +import java.time.Duration; import java.time.Instant; import java.util.Collection; import java.util.concurrent.CompletableFuture; @@ -181,6 +184,63 @@ default ResultStream stream(StreamEventsRequest request, i */ CompletableFuture getSequenceAt(Instant timestamp); + + /** + * Schedule the given {@code event} to be published after given {@code triggerDuration}. The returned value can be + * used to cancel the schedule, or to reschedule the event to another time. + * + * @param triggerDuration the amount of time to wait to publish the event + * @param event the event to publish + * @return a token used to cancel the schedule + */ + default CompletableFuture scheduleEvent(Duration triggerDuration, Event event) { + return scheduleEvent(Instant.now().plus(triggerDuration), event); + } + + /** + * Schedule the given {@code event} to be published at given {@code scheduleTime}. The returned value can be used to + * cancel the schedule, or to reschedule the event to another time. + * + * @param scheduleTime The scheduleTime at which to publish the event + * @param event The event to publish + * @return a token used to cancel the schedule + */ + CompletableFuture scheduleEvent(Instant scheduleTime, Event event); + + /** + * Cancels the scheduled publication of an event for which the given {@code scheduleToken} was returned. + * + * @param scheduleToken the token provided when scheduling the event to be cancelled + * @return a future reference to the result of the instruction + */ + CompletableFuture cancelSchedule(String scheduleToken); + + /** + * Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it + * to be published after given {@code triggerDuration}. If no event is provided, the original event will be + * rescheduled. + * + * @param scheduleToken the token of the event to cancel + * @param triggerDuration the point amount of time to wait to publish the event + * @param event an optional new event to publish + * @return a future reference to the token for the new schedule + */ + default CompletableFuture reschedule(String scheduleToken, Duration triggerDuration, Event event) { + return reschedule(scheduleToken, Instant.now().plus(triggerDuration), event); + } + + /** + * Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it + * to be published at given {@code scheduleTime}. If no event is provided, the original event will be + * rescheduled. + * + * @param scheduleToken the token of the event to cancel + * @param scheduleTime the point in time to publish the event + * @param event an optional new event to publish + * @return a future reference to the token for the new schedule + */ + CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, Event event); + /** * Provides operations to interact with a Transaction to append events and a condition onto the Event Store. */ diff --git a/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java b/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java index ce9f876e..bc22b9f7 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java @@ -17,6 +17,8 @@ package io.axoniq.axonserver.connector.event.impl; import com.google.protobuf.Empty; +import io.axoniq.axonserver.connector.AxonServerException; +import io.axoniq.axonserver.connector.ErrorCategory; import io.axoniq.axonserver.connector.ResultStream; import io.axoniq.axonserver.connector.event.DcbEventChannel; import io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel; @@ -24,13 +26,17 @@ 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.InstructionAck; import io.axoniq.axonserver.grpc.control.ClientIdentification; import io.axoniq.axonserver.grpc.event.dcb.AddTagsRequest; import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse; import io.axoniq.axonserver.grpc.event.dcb.AppendEventsRequest; import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse; +import io.axoniq.axonserver.grpc.event.dcb.CancelScheduledEventRequest; import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition; +import io.axoniq.axonserver.grpc.event.dcb.DcbEventSchedulerGrpc; import io.axoniq.axonserver.grpc.event.dcb.DcbEventStoreGrpc; +import io.axoniq.axonserver.grpc.event.dcb.Event; import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest; import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse; import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest; @@ -41,6 +47,9 @@ import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse; import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsRequest; import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse; +import io.axoniq.axonserver.grpc.event.dcb.RescheduleEventRequest; +import io.axoniq.axonserver.grpc.event.dcb.ScheduleEventRequest; +import io.axoniq.axonserver.grpc.event.dcb.ScheduleToken; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest; @@ -75,6 +84,7 @@ public class DcbEventChannelImpl extends AbstractAxonServerChannel impleme private static final int BUFFER_SIZE = 512; private static final int REFILL_BATCH = 16; private final DcbEventStoreGrpc.DcbEventStoreStub eventStore; + private final DcbEventSchedulerGrpc.DcbEventSchedulerStub eventScheduler; private final ClientIdentification clientIdentification; private final Set> buffers = ConcurrentHashMap.newKeySet(); @@ -90,6 +100,7 @@ public DcbEventChannelImpl(ClientIdentification clientIdentification, AxonServerManagedChannel axonServerManagedChannel) { super(clientIdentification, executor, axonServerManagedChannel); this.eventStore = DcbEventStoreGrpc.newStub(axonServerManagedChannel); + this.eventScheduler = DcbEventSchedulerGrpc.newStub(axonServerManagedChannel); this.clientIdentification = clientIdentification; } @@ -236,6 +247,52 @@ public CompletableFuture removeTags(long sequence, Collectio return future; } + @Override + public CompletableFuture scheduleEvent(Instant scheduleTime, Event event) { + FutureStreamObserverresponseObserver = new FutureStreamObserver<>(new AxonServerException( + ErrorCategory.OTHER, + "An unknown error occurred while scheduling an Event. No response received from Server.", + "" + )); + + eventScheduler.scheduleEvent(ScheduleEventRequest.newBuilder() + .setEvent(event) + .setInstant(scheduleTime.toEpochMilli()) + .build(), responseObserver); + return responseObserver.thenApply(ScheduleToken::getToken); + } + + @Override + public CompletableFuture cancelSchedule(String token) { + FutureStreamObserver responseObserver = new FutureStreamObserver<>(new AxonServerException( + ErrorCategory.OTHER, + "An unknown error occurred while cancelling a scheduled Event. No response received from Server.", + "" + )); + + eventScheduler.cancelScheduledEvent(CancelScheduledEventRequest.newBuilder().setToken(token).build(), + responseObserver); + return responseObserver; + } + + @Override + public CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, Event event) { + FutureStreamObserverresponseObserver = new FutureStreamObserver<>(new AxonServerException( + ErrorCategory.OTHER, + "An unknown error occurred while rescheduling Event. No response received from Server.", + "" + )); + + eventScheduler.rescheduleEvent(RescheduleEventRequest.newBuilder() + .setToken(scheduleToken) + .setEvent(event) + .setInstant(scheduleTime.toEpochMilli()) + .build(), + responseObserver); + return responseObserver.thenApply(ScheduleToken::getToken); + } + + private static class AppendEventsTransactionImpl implements AppendEventsTransaction { private final StreamObserver stream; 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 dfae9223..dce8c905 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,21 +7,15 @@ import io.axoniq.axonserver.connector.AxonServerConnectionFactory; import io.axoniq.axonserver.connector.ResultStream; import io.axoniq.axonserver.connector.ResultStreamPublisher; -import io.axoniq.axonserver.connector.impl.ServerAddress; import io.axoniq.axonserver.connector.event.DcbEventChannel; +import io.axoniq.axonserver.connector.impl.ServerAddress; 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.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.AddTagsResponse; -import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest; import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse; import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest; @@ -36,11 +30,9 @@ 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; -import java.util.Arrays; import java.util.List; import java.util.Set; import java.util.UUID; @@ -917,6 +909,57 @@ void getSequenceAtBetweenEvents() { } } + @Test + void scheduleEvent() { + DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); + long head = dcbEventChannel.head().join().getSequence(); + Event event = anEvent(aString(), "scheduledEvent"); + dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event) + .join(); + StepVerifier.create(streamFlux(head).take(1)) + .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("scheduledEvent")) + .verifyComplete(); + } + + @Test + void cancelScheduledEvent() { + DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); + long head = dcbEventChannel.head().join().getSequence(); + Event event = anEvent(aString(), "scheduledEvent"); + String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(1), event) + .join(); + dcbEventChannel.cancelSchedule(token).join(); + StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(2))) + .verifyComplete(); + } + + @Test + void rescheduleEvent() { + DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); + long head = dcbEventChannel.head().join().getSequence(); + Event event = anEvent(aString(), "rescheduledEvent"); + String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(30), event) + .join(); + dcbEventChannel.reschedule(token, Duration.ofSeconds(2), null).join(); + StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(3))) + .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent")) + .verifyComplete(); + } + + @Test + void rescheduleAndReplaceEvent() { + DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); + long head = dcbEventChannel.head().join().getSequence(); + Event event = anEvent(aString(), "scheduledEvent"); + String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event) + .join(); + Event replaceEvent = anEvent(aString(), "rescheduledEvent"); + dcbEventChannel.reschedule(token, Duration.ofSeconds(2), replaceEvent).join(); + StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(6))) + .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent")) + .verifyComplete(); + } + private Flux sourceFlux(long start) { DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); return fluxStream(() -> dcbEventChannel.source(sourceEventsRequest(start))); From 90580b1f532239984105be35740321ae6bd06b1d Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 5 Sep 2025 10:27:32 +0200 Subject: [PATCH 2/3] process review comments --- .../connector/event/DcbEventChannel.java | 3 ++- .../event/impl/DcbEventChannelImpl.java | 24 +++++++++---------- .../AbstractAxonServerIntegrationTest.java | 2 +- .../connector/event/dcb/DcbEndToEndTest.java | 9 ++++--- 4 files changed, 18 insertions(+), 20 deletions(-) diff --git a/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java b/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java index f48346d5..eee403b4 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/DcbEventChannel.java @@ -38,6 +38,7 @@ import java.time.Instant; import java.util.Collection; import java.util.concurrent.CompletableFuture; +import javax.annotation.Nullable; /** * Communication channel for Event related interaction with Axon Server based on the Dynamic Consistency Boundaries @@ -239,7 +240,7 @@ default CompletableFuture reschedule(String scheduleToken, Duration trig * @param event an optional new event to publish * @return a future reference to the token for the new schedule */ - CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, Event event); + CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event); /** * Provides operations to interact with a Transaction to append events and a condition onto the Event Store. diff --git a/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java b/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java index bc22b9f7..fcfd2453 100644 --- a/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java +++ b/src/main/java/io/axoniq/axonserver/connector/event/impl/DcbEventChannelImpl.java @@ -59,8 +59,6 @@ import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.grpc.stub.StreamObserver; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.time.Instant; import java.util.Collection; @@ -69,6 +67,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nullable; /** * {@link DcbEventChannel} implementation, serving as the event connection between Axon Server and a client @@ -79,8 +78,6 @@ */ public class DcbEventChannelImpl extends AbstractAxonServerChannel implements DcbEventChannel { - private static final Logger LOGGER = LoggerFactory.getLogger(DcbEventChannelImpl.class); - private static final int BUFFER_SIZE = 512; private static final int REFILL_BATCH = 16; private final DcbEventStoreGrpc.DcbEventStoreStub eventStore; @@ -250,7 +247,7 @@ public CompletableFuture removeTags(long sequence, Collectio @Override public CompletableFuture scheduleEvent(Instant scheduleTime, Event event) { FutureStreamObserverresponseObserver = new FutureStreamObserver<>(new AxonServerException( - ErrorCategory.OTHER, + ErrorCategory.INSTRUCTION_ACK_ERROR, "An unknown error occurred while scheduling an Event. No response received from Server.", "" )); @@ -265,7 +262,7 @@ public CompletableFuture scheduleEvent(Instant scheduleTime, Event event @Override public CompletableFuture cancelSchedule(String token) { FutureStreamObserver responseObserver = new FutureStreamObserver<>(new AxonServerException( - ErrorCategory.OTHER, + ErrorCategory.INSTRUCTION_ACK_ERROR, "An unknown error occurred while cancelling a scheduled Event. No response received from Server.", "" )); @@ -276,19 +273,20 @@ public CompletableFuture cancelSchedule(String token) { } @Override - public CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, Event event) { + public CompletableFuture reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event) { FutureStreamObserverresponseObserver = new FutureStreamObserver<>(new AxonServerException( - ErrorCategory.OTHER, + ErrorCategory.INSTRUCTION_ACK_ERROR, "An unknown error occurred while rescheduling Event. No response received from Server.", "" )); - eventScheduler.rescheduleEvent(RescheduleEventRequest.newBuilder() + RescheduleEventRequest.Builder builder = RescheduleEventRequest.newBuilder() .setToken(scheduleToken) - .setEvent(event) - .setInstant(scheduleTime.toEpochMilli()) - .build(), - responseObserver); + .setInstant(scheduleTime.toEpochMilli()); + if (event != null) { + builder.setEvent(event); + } + eventScheduler.rescheduleEvent(builder.build(), responseObserver); return responseObserver.thenApply(ScheduleToken::getToken); } diff --git a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java index 9a2802f2..a67ccb0a 100644 --- a/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java +++ b/src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java @@ -85,7 +85,7 @@ private static String getDockerImageName() { protected static Proxy axonServerProxy; protected static ServerAddress axonServerAddress; - private static ServerAddress axonServerHttpPort; + protected static ServerAddress axonServerHttpPort; protected String axonServerVersion; @BeforeAll 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 dce8c905..74640800 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 @@ -63,11 +63,6 @@ class DcbEndToEndTest extends AbstractAxonServerIntegrationTest { */ private static final boolean LOCAL = false; - /** - * HTTP port for local Axon Server instance - */ - private static ServerAddress axonServerHttpPort; - /** * Static initialization for local mode */ @@ -910,6 +905,7 @@ void getSequenceAtBetweenEvents() { } @Test + @Disabled("Enable when scheduling is supported in AxonServer") void scheduleEvent() { DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); long head = dcbEventChannel.head().join().getSequence(); @@ -922,6 +918,7 @@ void scheduleEvent() { } @Test + @Disabled("Enable when scheduling is supported in AxonServer") void cancelScheduledEvent() { DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); long head = dcbEventChannel.head().join().getSequence(); @@ -934,6 +931,7 @@ void cancelScheduledEvent() { } @Test + @Disabled("Enable when scheduling is supported in AxonServer") void rescheduleEvent() { DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); long head = dcbEventChannel.head().join().getSequence(); @@ -947,6 +945,7 @@ void rescheduleEvent() { } @Test + @Disabled("Enable when scheduling is supported in AxonServer") void rescheduleAndReplaceEvent() { DcbEventChannel dcbEventChannel = connection.dcbEventChannel(); long head = dcbEventChannel.head().join().getSequence(); From 2fddea9c5bb8c48f1451dc3497abb33811fe0010 Mon Sep 17 00:00:00 2001 From: Marc Date: Fri, 5 Sep 2025 11:57:55 +0200 Subject: [PATCH 3/3] add wait after recreating context to make tests more stable. --- .../axonserver/connector/testutils/AxonServerUtils.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/test/java/io/axoniq/axonserver/connector/testutils/AxonServerUtils.java b/src/test/java/io/axoniq/axonserver/connector/testutils/AxonServerUtils.java index e2aaab65..92b8ae39 100644 --- a/src/test/java/io/axoniq/axonserver/connector/testutils/AxonServerUtils.java +++ b/src/test/java/io/axoniq/axonserver/connector/testutils/AxonServerUtils.java @@ -47,6 +47,7 @@ public static void purgeEventsFromAxonServer(String context, String hostname, in throws IOException { deleteContext(context, hostname, port); createContext(context, hostname, port, dcbContext); + waitFor(1000); } public static void deleteContext(String context, String hostname, int port) throws IOException { @@ -182,4 +183,12 @@ private static void checkContextsCondition(String hostname, int port, throw new RuntimeException(e); } } + + private static void waitFor(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } }