Persistent streams #347

MGathier · 2024-02-15T10:07:28Z

No description provided.

abuijze · 2024-02-23T08:58:36Z

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

+    private final int segments;
+    private final String sequencingPolicyName;
+    private final List<String> sequencingPolicyParameters;
+    private final int initialPosition;


Position should be a long

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

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

abuijze · 2024-02-23T09:16:28Z

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

+     * or to only sent progress information after a number of processed events.
+     * @param token the last processed token for this segment
+     */
+    void acknowledge(long token);


What happens when the client cannot send the ack to the server, for example when the connection is lost? Should we return a CompletableFuture to... ehhh... ack the ack?

add testcases

abuijze · 2024-03-14T10:04:34Z

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

+
+import java.util.List;
+
+public class PersistedStreamProperties {


Typo in the classname. Should be PersistentStreamProperties

abuijze · 2024-03-14T10:05:07Z

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

+     * created with the properties specified in {@code creationProperties}.
+     * @param streamId the unique identification of a persistent stream
+     * @param creationProperties properties to initialize the persistent stream if it does not exist yet
+     * @return a PersistedStream streaming events per segment


PersistedStream -> PersistentStream

abuijze · 2024-03-14T10:05:52Z

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

+
+
+    /**
+     * Deletes a persistent stream.


Perhaps explicitly mention that this command will also return normally if the stream does not exist (if that's the way it actually behaves, of course)

abuijze · 2024-03-14T10:06:48Z

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

+     * @param segments the requested number of segments for the persistent stream
+     * @return a CompletableFuture that completes when the persistent stream is updated
+     */
+    CompletableFuture<Void> setPersistentStreamSegments(String streamId, Integer segments);


Why is this an Integer, rather than an int? Is null an option somehow?

abuijze · 2024-03-14T10:08:10Z

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

+     * Returns a list of persistent streams with the connected client per segment.
+     * @return a list of persistent streams
+     */
+    CompletableFuture<List<StreamConnections>> persistentStreamConnections();


Couldn't this become potentially very long? What about returning the connections for a given streamId? Then the persistentStreams() call can be used to retrieve all ID's, and the details can be fetched for those that are relevant.

abuijze · 2024-03-14T10:09:12Z

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

+    private final long initialPosition;
+    private final String filter;
+
+    public PersistedStreamProperties(String streamName, int segments, String sequencingPolicyName,


Javadoc is missing on this class.

Is 0 a valid value for segments? If not, would be good to validate that.

abuijze · 2024-03-14T10:16:35Z

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

+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.LongConsumer;
+
+public class BufferedPersistentStreamSegment


Is there a specific reason this class doesn't extend the AbstractBufferedStream?

abuijze · 2024-03-14T10:26:56Z

src/test/java/io/axoniq/axonserver/connector/event/impl/PersistentEventChannelImplTest.java

+
+    @BeforeEach
+    void setup() throws ExecutionException, InterruptedException {
+        AxonServerConnectionFactory connectionFactory = AxonServerConnectionFactory.forClient("demo",


This test seems to assume AxonServer is running on port 8124 to run. Could we use testcontainers instead?

abuijze · 2024-03-14T10:28:20Z

src/test/java/io/axoniq/axonserver/connector/event/impl/PersistentEventChannelImplTest.java

+                                       .build()).get(1, TimeUnit.SECONDS);
+
+        long last = eventChannel.getLastToken().get();
+        Thread.sleep(2000);


There is an assertWithin in the AssertUtils class which would prevent the need for arbitrary sleeps.

There don't seem to be any assertions in this test either.

let client control flow of events for a persistent stream.

smcvb

Bunch of remarks that would clean up the new code before we merge it.

pom.xml

smcvb · 2024-04-25T14:41:42Z

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

+
+    /**
+     * Opens the persistent stream identified by {@code streamId}. Fails if the stream does not exist.
+     * @param streamId the unique identification of a persistent stream


Not all parameters are named in the JavaDoc.

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

src/test/java/io/axoniq/axonserver/connector/event/impl/PersistentStreamImplTest.java

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

smcvb · 2024-04-25T15:00:16Z

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

+        return InitializationProperties.newBuilder()
+                                       .setInitialPosition(creationProperties.initialPosition())
+                                       .setSegments(creationProperties.segments())
+                                       .setStreamName(nonNullOrDefault(creationProperties.streamName(),""))


An empty String as the Stream name seems rather useless to me. Wouldn't a shortened UUID better fit the bill if nothing's provided?

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

# Conflicts: # pom.xml

update callback signature to provide the PersistentStreamSegment instead of just the segment number.

…fter a close request wait for segments to complete the active batch on stopping the application

…2024.1

smcvb

My concerns have been addressed, hence I'm approving this pull request.

MGathier added 6 commits February 6, 2024 16:25

first implementation of persistent stream operations

e9ef61e

first implementation of persistent stream operations

aad728b

rename persisted streams to persistent streams

9e978db

rename classes in EventChannel interface

a52491f

small cleanup in the persistent stream API

cc997e2

more javadoc

a38a8e8

MGathier added this to the Axon Server Connector for Java 2024.0.0 milestone Feb 15, 2024

MGathier requested review from abuijze, smcvb, m1l4n54v1c and saratry February 15, 2024 10:07

MGathier self-assigned this Feb 15, 2024

rename service and messages in persistent-streams.proto.

fbd97e6

abuijze reviewed Feb 23, 2024

View reviewed changes

MGathier added 4 commits February 27, 2024 10:08

process review comments

464eb9d

add testcases

process review comments

378cafa

add testcases

remove invalid test

9efbb31

Merge remote-tracking branch 'origin/master' into persisted-streams

3157e5a

abuijze requested changes Mar 14, 2024

View reviewed changes

MGathier added 3 commits March 18, 2024 11:40

process review comments.

c78cd44

let client control flow of events for a persistent stream.

process review comments.

d290f4e

let client control flow of events for a persistent stream.

fix compilation errors

9ebfd91

smcvb added Type: Feature Status: In Progress Priority 1: Must labels Apr 25, 2024

smcvb requested changes Apr 25, 2024

View reviewed changes

MGathier added 4 commits April 26, 2024 09:11

Merge remote-tracking branch 'origin/master' into persisted-streams

d3dc847

# Conflicts: # pom.xml

resolve review remarks

aaf43c9

cancel of segment must not cancel the gRPC request

971bab5

add operation to notify server when a segment has an error.

2c0949b

update callback signature to provide the PersistentStreamSegment instead of just the segment number.

MGathier added 3 commits June 4, 2024 11:38

add support for dead letter queues

1f6498d

let segment send a marker when it is done processing the last batch a…

69035f3

…fter a close request wait for segments to complete the active batch on stopping the application

code cleanup

134a48a

gklijs mentioned this pull request Jun 10, 2024

Persistent streams AxonFramework/AxonFramework#3030

Merged

MGathier added 6 commits June 27, 2024 13:51

reset persistent stream position

56c7f33

reset persistent stream position - small change in interface

ebb0f0e

reset persistent stream position - missing documentation

34e08ee

reset persistent stream position - change in API

e503d2e

remove wrapper from ResetStreamConfiguration

75b8f52

update version to 2024.1.0-SNAPSHOT as this is linked to API version …

7ffc1c7

…2024.1

MGathier merged commit c270a48 into master Jul 2, 2024
3 of 5 checks passed

MGathier deleted the persisted-streams branch July 2, 2024 13:12

smcvb added Status: Resolved and removed Status: In Progress labels Jul 16, 2024

smcvb reviewed Jul 17, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Persistent streams #347

Persistent streams #347

MGathier commented Feb 15, 2024

abuijze Feb 23, 2024

abuijze Feb 23, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

abuijze Mar 14, 2024

smcvb left a comment

smcvb Apr 25, 2024

smcvb Apr 25, 2024

smcvb left a comment


		import java.util.List;

		public class PersistedStreamProperties {

Persistent streams #347

Persistent streams #347

Conversation

MGathier commented Feb 15, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smcvb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smcvb left a comment

Choose a reason for hiding this comment