Streaming query support #112

schananas · 2021-11-09T09:38:33Z

Adds few adjustemnts and fixes for better streaming query support

CLAassistant · 2021-11-09T09:38:39Z

All committers have signed the CLA.

schananas · 2021-11-09T09:52:32Z

Integration tests to be done

src/main/java/io/axoniq/axonserver/connector/query/QueryHandler.java

abuijze · 2021-12-09T11:58:44Z

src/main/java/io/axoniq/axonserver/connector/query/impl/MultiFlowControl.java

+
+    @Override
+    public void request(long requested) {
+        delegates.parallelStream()


Wouldn't this cause a "request 1" to lead to more than 1 message being sent? Is this acceptable or should we buffer the responses and deplete the buffer before asking more from upstream?

Ah, you are right! We could also do a round-robin without the buffer - no necessary invocations. What do you think?

abuijze · 2021-12-09T12:24:39Z

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

+        result.complete();
+    }
+
+    private void flowControlQuery(QueryProviderInbound flowControl, ReplyChannel<QueryProviderOutbound> result) {


to avoid confusion, I suggest renaming. For example: handleFlowControlRequest

abuijze · 2021-12-09T12:25:09Z

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

@@ -191,6 +193,18 @@ public void complete() {
        result.complete();
    }

+    private void completeStreamingQuery(QueryProviderInbound complete, ReplyChannel<QueryProviderOutbound> result) {


In line with comment below suggestion: handleCancelRequest

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

abuijze · 2021-12-09T12:39:37Z

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

+        MultiFlowControl multiFlowControl = new MultiFlowControl();
+        Runnable removeQuery = () -> queriesInProgress.remove(query.getMessageIdentifier());
+        QueryInProgress queryInProgress = new QueryInProgress(removeQuery, multiFlowControl);
+        queriesInProgress.putIfAbsent(query.getMessageIdentifier(), queryInProgress);


The use of putIfAbsent suggests that the given identifier may already exist. In that case, wouldn't we want to use the already existing QueryInProgess instance? Of maybe reject the request altogether?

Semantically, I think we want to reject the request. If there is already the same query in progress, responses sent are valid, and responses to be sent are going to be valid. If you want a new result stream, issue a new query. If, however, receiving the same query twice is a result of a retry, rejecting it is also fine.

abuijze · 2021-12-09T12:42:26Z

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

+            return closeHandler;
+        }
+
+        public void complete() {


Suggest: rename to cancel()

Instead of completing normally, perhaps complete with a QueryCancelledException instead?

Not sure whether we want to complete exceptionally here. Having a receiver cancelling the stream doesn't seem like an exceptional case but rather a regular one.

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

sonarcloud · 2021-12-13T12:26:08Z

Kudos, SonarCloud Quality Gate passed!

0 Bugs
0 Vulnerabilities
0 Security Hotspots
3 Code Smells

78.5% Coverage
0.0% Duplication

# Conflicts: # pom.xml

Added test for exception throwing FlowControl.

smcvb

Implementation wise I approve this PR, as I assume is known through the virtual reviews we've done. I've got plenty of JavaDoc suggestions as always, however.

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

src/main/java/io/axoniq/axonserver/connector/query/QueryChannel.java

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

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

...a/io/axoniq/axonserver/connector/impl/buffer/FlowControlledDisposableReadonlyBufferTest.java

src/test/java/io/axoniq/axonserver/connector/impl/buffer/RoundRobinMultiReadonlyBufferTest.java

src/main/java/io/axoniq/axonserver/connector/query/impl/QueryChannelImpl.java

smcvb · 2022-03-11T12:27:23Z

Also, don't forget to take a look at Sonar. It marks one bug and four code smells.

smcvb

My concerns have been addressed, hence approving.

sonarcloud · 2022-03-24T14:47:14Z

Kudos, SonarCloud Quality Gate passed!

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

0.0% Coverage
0.0% Duplication

m1l4n54v1c and others added 4 commits October 28, 2021 11:30

Added a way to register a listener when query completes.

0b6c917

Add flow control messages and listeners for query channel

0372f2c

Fix peek() in FlowControlledBuffer to work as described in doc

14d1c8e

add logger for unknown requests

1fbe25d

schananas added 2 commits November 9, 2021 10:45

fix javadoc and todo remove

a1a6049

fix javadoc and todo remove

2231e19

schananas requested review from smcvb and abuijze November 9, 2021 09:47

schananas marked this pull request as draft November 9, 2021 10:00

add registration for query complete and query flow control

9e3cacd

smcvb added Priority 1: Must Status: In Progress Type: Feature labels Nov 12, 2021

smcvb added this to the Release 4.6.0 milestone Nov 12, 2021

smcvb assigned schananas Nov 12, 2021

m1l4n54v1c added 2 commits November 24, 2021 11:59

Fixed Query Flow Control typo.

1c082cd

QueryHandler now returns a handle for FlowControl.

4509600

abuijze requested changes Dec 9, 2021

View reviewed changes

Code review remarks.

c412f06

m1l4n54v1c added 6 commits December 14, 2021 10:16

Added test for MultiFlowControl with a single handler.

21a54a8

Added copyright to new files.

487c650

Added buffering flow control capabilities to query functionality.

4151402

Merge branch 'feature/adminApi/getEventProcessors' into streaming-query

bbd632b

# Conflicts: # pom.xml

Merge branch 'feature/adminApi/getEventProcessors' into streaming-query

31d060e

Merge branch 'master' into streaming-query

55c7f40

m1l4n54v1c self-assigned this Feb 22, 2022

m1l4n54v1c marked this pull request as ready for review February 22, 2022 10:49

m1l4n54v1c added 2 commits March 1, 2022 14:24

code review: API changes

5ce2db7

Added @SInCE.

d412ba5

Added test for exception throwing FlowControl.

smcvb requested changes Mar 11, 2022

View reviewed changes

m1l4n54v1c added 2 commits March 21, 2022 17:31

Review comments.

2a23d5b

Sonar comments.

1f91110

smcvb approved these changes Mar 24, 2022

View reviewed changes

Adepted to the latest API changes.

afdfd2e

m1l4n54v1c merged commit 731f853 into master Mar 24, 2022

m1l4n54v1c deleted the streaming-query branch March 24, 2022 14:44

smcvb added Status: Resolved and removed Status: In Progress labels Mar 25, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Streaming query support #112

Streaming query support #112

schananas commented Nov 9, 2021

CLAassistant commented Nov 9, 2021 •

edited

schananas commented Nov 9, 2021

abuijze Dec 9, 2021

m1l4n54v1c Dec 9, 2021

abuijze Dec 9, 2021

abuijze Dec 9, 2021

abuijze Dec 9, 2021

m1l4n54v1c Dec 9, 2021

abuijze Dec 9, 2021

m1l4n54v1c Dec 13, 2021

sonarcloud bot commented Dec 13, 2021

smcvb left a comment

smcvb commented Mar 11, 2022

smcvb left a comment

sonarcloud bot commented Mar 24, 2022

Streaming query support #112

Streaming query support #112

Conversation

schananas commented Nov 9, 2021

CLAassistant commented Nov 9, 2021 • edited

schananas commented Nov 9, 2021

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

sonarcloud bot commented Dec 13, 2021

smcvb left a comment

Choose a reason for hiding this comment

smcvb commented Mar 11, 2022

smcvb left a comment

Choose a reason for hiding this comment

sonarcloud bot commented Mar 24, 2022

CLAassistant commented Nov 9, 2021 •

edited