You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "mmodzelewski (via GitHub)" <gi...@apache.org> on 2026/08/07 19:04:10 UTC
[PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
mmodzelewski opened a new pull request, #3841:
URL: https://github.com/apache/iggy/pull/3841
The Java SDK spoke only the legacy TCP framing, so it could not
talk to server-ng, which frames every request in the VSR
(Viewstamped Replication) consensus protocol.
The Netty pipeline now encodes requests with VSR headers from a
per-connection consensus session (client id, request counter,
fence epoch) and decodes VSR replies, replacing the legacy frame
codec. The server binds the session to one transport connection,
so the pool pins a single channel and the connectionPoolSize
builder option is gone. Transient denials replay the same encoded
frame so the server dedup sees one request id: not-committed for
the full 30s budget (outcome unknown), not-accepted for 2s only
(refused outright, typically a demoted primary). A failed pool
acquire starts a background redial that alternates the current
node with the seed and replays the builder credentials.
The VSR broker routes explicit partitions only, so balanced and
key-based sends resolve to a concrete partition client-side
(round-robin cursor, xxh32(key) % count) and group polls pick the
partition from the member's assignment, synced through the new
consumer-group SYNC command and fenced by generation, matching
the Rust SDK. Cached routing facts refresh past a 5s staleness
budget. sendMessages returns the commit confirmation (partition
and base offset) instead of Void.
Java CI, BDD, and examples lanes run against iggy-server-ng with
the vsr feature; the SDK version bumps to 0.9.0-SNAPSHOT. On the
server side, a login whose credentials were rejected no longer
falls through to the malformed-payload rejection but reports
InvalidCredentials.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski merged PR #3841:
URL: https://github.com/apache/iggy/pull/3841
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski commented on PR #3841:
URL: https://github.com/apache/iggy/pull/3841#issuecomment-5233016993
@slbotbm Thanks for the review. I've addressed all the comments.
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3841:
URL: https://github.com/apache/iggy/pull/3841#issuecomment-5221006105
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3841?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 75.46%. Comparing base ([`02c24b3`](https://app.codecov.io/gh/apache/iggy/commit/02c24b399100a168c610ce0994e6c79d384a71cc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`80490c7`](https://app.codecov.io/gh/apache/iggy/commit/80490c7d96cf39faf61e81e1a8149973f092f217?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3841 +/- ##
============================================
- Coverage 75.89% 75.46% -0.44%
+ Complexity 1046 1030 -16
============================================
Files 1337 1335 -2
Lines 169995 169636 -359
Branches 142348 142276 -72
============================================
- Hits 129025 128013 -1012
- Misses 37168 37825 +657
+ Partials 3802 3798 -4
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.79% <ø> (+<0.01%)` | :arrow_up: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.28% <ø> (-0.40%)` | :arrow_down: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.26% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `88.70% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.97% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.68% <ø> (-6.53%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3841/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.08% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3841?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/server-ng/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL2Rpc3BhdGNoLnJz) | `40.80% <ø> (+0.03%)` | :arrow_up: |
| [...apache/iggy/connector/flink/source/IggySource.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fexternal-processors%2Figgy-connector-flink%2Figgy-connector-library%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fconnector%2Fflink%2Fsource%2FIggySource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2V4dGVybmFsLXByb2Nlc3NvcnMvaWdneS1jb25uZWN0b3ItZmxpbmsvaWdneS1jb25uZWN0b3ItbGlicmFyeS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaWdneS9jb25uZWN0b3IvZmxpbmsvc291cmNlL0lnZ3lTb3VyY2UuamF2YQ==) | `0.00% <ø> (ø)` | |
| [...apache/iggy/client/async/ConsumerGroupsClient.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2FConsumerGroupsClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy9Db25zdW1lckdyb3Vwc0NsaWVudC5qYXZh) | `0.00% <ø> (ø)` | |
| [...a/org/apache/iggy/client/async/MessagesClient.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2FMessagesClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy9NZXNzYWdlc0NsaWVudC5qYXZh) | `0.00% <ø> (ø)` | |
| [...ache/iggy/client/async/tcp/AsyncIggyTcpClient.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FAsyncIggyTcpClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvQXN5bmNJZ2d5VGNwQ2xpZW50LmphdmE=) | `67.33% <ø> (-1.06%)` | :arrow_down: |
| [...gy/client/async/tcp/AsyncIggyTcpClientBuilder.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FAsyncIggyTcpClientBuilder.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvQXN5bmNJZ2d5VGNwQ2xpZW50QnVpbGRlci5qYXZh) | `90.76% <ø> (-0.79%)` | :arrow_down: |
| [...ache/iggy/client/async/tcp/AsyncTcpConnection.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FAsyncTcpConnection.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvQXN5bmNUY3BDb25uZWN0aW9uLmphdmE=) | `71.03% <ø> (-2.54%)` | :arrow_down: |
| [...iggy/client/async/tcp/ConsumerGroupsTcpClient.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FConsumerGroupsTcpClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvQ29uc3VtZXJHcm91cHNUY3BDbGllbnQuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...pache/iggy/client/async/tcp/IggyAuthenticator.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FIggyAuthenticator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvSWdneUF1dGhlbnRpY2F0b3IuamF2YQ==) | `90.90% <ø> (+5.72%)` | :arrow_up: |
| [...pache/iggy/client/async/tcp/MessagesTcpClient.java](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree&filepath=foreign%2Fjava%2Fjava-sdk%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Figgy%2Fclient%2Fasync%2Ftcp%2FMessagesTcpClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9qYXZhL2phdmEtc2RrL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9pZ2d5L2NsaWVudC9hc3luYy90Y3AvTWVzc2FnZXNUY3BDbGllbnQuamF2YQ==) | `97.29% <ø> (-0.14%)` | :arrow_down: |
| ... and [9 more](https://app.codecov.io/gh/apache/iggy/pull/3841?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [52 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3841/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3841:
URL: https://github.com/apache/iggy/pull/3841#discussion_r3740610881
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java:
##########
@@ -177,13 +180,45 @@ public static ConsumerGroup readConsumerGroup(ByteBuf response) {
return new ConsumerGroup(groupId, name, partitionsCount, membersCount);
}
+ public static ConsumerGroupAssignment readConsumerGroupAssignment(ByteBuf response) {
+ // The generation is a monotonic rebalance counter compared only for
+ // equality, so reading the u64 as a signed long is safe.
+ var generation = response.readLongLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ List<Long> partitions = new ArrayList<>(toInt(partitionsCount));
+ for (long i = 0; i < partitionsCount; i++) {
+ partitions.add(response.readUnsignedIntLE());
+ }
+ return new ConsumerGroupAssignment(generation, partitions);
Review Comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3841:
URL: https://github.com/apache/iggy/pull/3841#discussion_r3740591462
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java:
##########
@@ -193,100 +234,225 @@ public CompletableFuture<ByteBuf> send(CommandCode commandCode, ByteBuf payload)
}
public CompletableFuture<ByteBuf> send(int commandCode, ByteBuf payload) {
+ if (isLoginCode(commandCode) && authenticated) {
+ return logoutThenLogin(commandCode, payload);
+ }
captureLoginPayloadIfNeeded(commandCode, payload);
CompletableFuture<ByteBuf> responseFuture = new CompletableFuture<>();
CompletableFuture<ByteBuf> callerFuture = new CompletableFuture<>();
channelPool.acquire().addListener((FutureListener<Channel>) f -> {
if (!f.isSuccess()) {
payload.release();
+ notifyConnectionFailure(f.cause());
callerFuture.completeExceptionally(mapAcquireException(f.cause()));
return;
}
+ dispatchOnChannel(f.getNow(), commandCode, payload, responseFuture, callerFuture);
+ });
- Channel channel = f.getNow();
- boolean isLoginCommand = (commandCode == CommandCode.User.LOGIN.getValue()
- || commandCode == CommandCode.PersonalAccessToken.LOGIN.getValue());
- boolean requiresAuth = !isLoginCommand
- && commandCode != CommandCode.System.PING.getValue()
- && commandCode != CommandCode.System.GET_STATS.getValue();
+ return callerFuture;
+ }
- responseFuture.whenComplete((response, error) -> {
- try {
- handlePostResponse(channel, commandCode, isLoginCommand, error);
- } catch (RuntimeException bookkeepingError) {
- log.error("Post-response bookkeeping failed: {}", bookkeepingError.getMessage());
- }
- if (error != null) {
- callerFuture.completeExceptionally(error);
- } else {
- callerFuture.complete(response);
- }
- });
+ private void dispatchOnChannel(
+ Channel channel,
+ int commandCode,
+ ByteBuf payload,
+ CompletableFuture<ByteBuf> responseFuture,
+ CompletableFuture<ByteBuf> callerFuture) {
+ boolean isLoginCommand = isLoginCode(commandCode);
+ boolean requiresAuth = !isLoginCommand && requiresAuthentication(commandCode);
- CompletableFuture<Void> authStep;
- if (!requiresAuth) {
- authStep = CompletableFuture.completedFuture(null);
- } else if (!authenticated) {
+ responseFuture.whenComplete((response, error) -> {
+ try {
+ handlePostResponse(channel, commandCode, isLoginCommand, error);
+ } catch (RuntimeException bookkeepingError) {
+ log.error("Post-response bookkeeping failed: {}", bookkeepingError.getMessage());
+ }
+ if (error != null) {
+ callerFuture.completeExceptionally(error);
+ } else {
+ callerFuture.complete(response);
+ }
+ });
+
+ CompletableFuture<Void> authStep;
+ if (!requiresAuth) {
+ authStep = CompletableFuture.completedFuture(null);
+ } else if (!authenticated) {
+ payload.release();
+ responseFuture.completeExceptionally(new IggyNotConnectedException("Not authenticated, call login first"));
+ return;
+ } else {
+ ByteBuf loginPayloadCopy = getLoginPayloadCopy();
+ if (loginPayloadCopy == null) {
payload.release();
responseFuture.completeExceptionally(
new IggyNotConnectedException("Not authenticated, call login first"));
return;
- } else {
- ByteBuf loginPayloadCopy = getLoginPayloadCopy();
- if (loginPayloadCopy == null) {
- payload.release();
- responseFuture.completeExceptionally(
- new IggyNotConnectedException("Not authenticated, call login first"));
- return;
- }
- authStep = IggyAuthenticator.ensureAuthenticated(
- channel, loginPayloadCopy, loginCommandCode, authGeneration);
}
+ authStep = IggyAuthenticator.ensureAuthenticated(
+ channel, loginPayloadCopy, loginCommandCode, authGeneration, vsrEncoder);
+ }
- authStep.thenRun(() -> sendFrame(channel, payload, commandCode, responseFuture))
- .exceptionally(ex -> {
- responseFuture.completeExceptionally(ex);
- return null;
- });
+ authStep.whenComplete((ignored, authError) -> {
+ if (authError != null) {
+ payload.release();
+ responseFuture.completeExceptionally(authError);
+ return;
+ }
+ sendFrame(channel, payload, commandCode, responseFuture);
});
+ }
- return callerFuture;
+ /**
+ * A failed pool acquire on an established connection means the target
+ * node could not be (re)dialed; the listener lets the owning client run
+ * its redial strategy while the failed request surfaces to its caller.
+ */
+ private void notifyConnectionFailure(Throwable cause) {
+ try {
+ connectionFailureListener.accept(cause);
+ } catch (RuntimeException listenerError) {
+ log.warn("Connection failure listener threw: {}", listenerError.getMessage());
+ }
}
private static Throwable mapAcquireException(Throwable cause) {
if (cause instanceof IllegalStateException) {
return new IggyNotConnectedException("Connection pool is closed");
}
+ if (cause instanceof TimeoutException) {
+ return new IggyTimeoutException("Timed out acquiring a connection from the pool", cause);
+ }
return cause;
}
+ /**
+ * A Register on an already-bound VSR connection is answered with a replay
+ * of the original register reply, while the client has re-armed a fresh
+ * identity; its reset request counter would then collide with the
+ * server's dedup table and mutations would be silently swallowed. Unbind
+ * first, then login fresh.
+ */
+ private CompletableFuture<ByteBuf> logoutThenLogin(int commandCode, ByteBuf payload) {
+ return send(CommandCode.User.LOGOUT.getValue(), Unpooled.EMPTY_BUFFER)
+ .handle((logoutResponse, logoutError) -> {
+ if (logoutResponse != null) {
+ logoutResponse.release();
+ }
+ return null;
+ })
+ .thenCompose(ignored -> send(commandCode, payload));
+ }
+
+ private static boolean isLoginCode(int commandCode) {
+ return commandCode == CommandCode.User.LOGIN.getValue()
+ || commandCode == CommandCode.PersonalAccessToken.LOGIN.getValue();
+ }
+
+ /**
+ * Ping and cluster metadata are the only sessionless bootstrap commands.
+ * Cluster metadata must be available before Register so a VSR client can
+ * select the leader; every other non-login command requires a bound
+ * session.
+ */
+ private static boolean requiresAuthentication(int commandCode) {
+ return !isAllowedBeforeAuthentication(commandCode);
+ }
+
+ private static boolean isAllowedBeforeAuthentication(int commandCode) {
+ return commandCode == CommandCode.System.PING.getValue()
+ || commandCode == CommandCode.System.GET_CLUSTER_METADATA.getValue();
+ }
+
private void sendFrame(
Channel channel, ByteBuf payload, int commandCode, CompletableFuture<ByteBuf> responseFuture) {
try {
- IggyResponseHandler handler = channel.pipeline().get(IggyResponseHandler.class);
+ VsrResponseHandler handler = channel.pipeline().get(VsrResponseHandler.class);
if (handler == null) {
- throw new IggyClientException("Channel missing IggyResponseHandler");
+ throw new IggyClientException("Channel missing VsrResponseHandler");
}
- handler.enqueueRequest(responseFuture);
- ByteBuf frame = IggyFrameEncoder.encode(channel.alloc(), commandCode, payload);
-
- channel.writeAndFlush(frame).addListener((ChannelFutureListener) future -> {
- if (!future.isSuccess()) {
- log.error("Failed to send frame: {}", future.cause().getMessage());
- responseFuture.completeExceptionally(future.cause());
- } else {
- log.trace("Frame sent successfully to {}", channel.remoteAddress());
- }
- });
+ ByteBuf frame = vsrEncoder.encode(channel.alloc(), commandCode, payload);
+ long nowNanos = System.nanoTime();
+ long deadlineNanos = nowNanos + TRANSIENT_RETRY_BUDGET.toNanos();
+ long notAcceptedDeadlineNanos =
+ isLoginCode(commandCode) ? deadlineNanos : nowNanos + NOT_ACCEPTED_RETRY_BUDGET.toNanos();
+ writeVsrFrame(channel, handler, frame, responseFuture, deadlineNanos, notAcceptedDeadlineNanos);
} catch (RuntimeException e) {
responseFuture.completeExceptionally(e);
} finally {
payload.release();
}
}
Review Comment:
When a former leader remains reachable after an election, it can reject writes with `TransientNotAccepted`. `AsyncTcpConnection` retries the same frame on the same channel until `notAcceptedDeadlineNanos` and then completes the request exceptionally. `AsyncIggyTcpClient.onConnectionFailure()` only handles connection exceptions from failed pool acquisition, so this server response never triggers leader discovery and a healthy but demoted node is never replaced. Every later metadata write repeats the same failure until the application explicitly logs in again. Consider handling this status like the Rust VSR transport by fetching the current leader, reconnecting, re-registering, and reissuing the request; reissuing it elsewhere is safe because the server guarantees that it was not admitted.
Also applies to `AsyncIggyTcpClient.java:448-462`
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
Review Comment:
The migrated client never starts a heartbeat task. With server-ng's `[heartbeat] enabled = true`, only `PING` updates a connection's heartbeat timestamp; ordinary polls and writes do not. The verifier evicts a consumer-group member after roughly `1.2 * heartbeat.interval`, so even an actively polling Java consumer loses its VSR session and group membership after a few seconds. Consider starting a periodic ping for the lifetime of each published connection, using a configurable interval, and canceling or restarting it on close and retarget.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java:
##########
@@ -177,13 +180,45 @@ public static ConsumerGroup readConsumerGroup(ByteBuf response) {
return new ConsumerGroup(groupId, name, partitionsCount, membersCount);
}
+ public static ConsumerGroupAssignment readConsumerGroupAssignment(ByteBuf response) {
+ // The generation is a monotonic rebalance counter compared only for
+ // equality, so reading the u64 as a signed long is safe.
+ var generation = response.readLongLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ List<Long> partitions = new ArrayList<>(toInt(partitionsCount));
+ for (long i = 0; i < partitionsCount; i++) {
+ partitions.add(response.readUnsignedIntLE());
+ }
+ return new ConsumerGroupAssignment(generation, partitions);
Review Comment:
Both the consumer-group assignment decoder here and the send-confirmation decoder at line 207 allocate an `ArrayList` directly from a wire-provided `u32` count before checking whether the body contains that many fixed-size entries. A short malformed response with a count near `Integer.MAX_VALUE` attempts a multi-gigabyte allocation and can terminate the JVM with `OutOfMemoryError`; `Math.toIntExact` only rejects values above the signed range. Consider bounding the count by `readableBytes() / 4` for assignments and `readableBytes() / 20` for confirmations before allocating the validated capacity.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp.vsr;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.apache.iggy.exception.IggyConnectionException;
+import org.apache.iggy.exception.IggyServerException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Correlates in-flight requests with responses in FIFO order, decodes VSR
+ * reply frames, and completes the pending request future with the command
+ * payload the typed deserializers expect. Mirrors {@code decode_response} in
+ * {@code core/sdk/src/vsr.rs}: eviction frames become typed errors, a nonzero
+ * header status is a pre-commit deny, and result-framed bodies have their
+ * committed result section stripped (or raised as the typed error).
+ */
+public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
+
+ private static final Logger log = LoggerFactory.getLogger(VsrResponseHandler.class);
+
+ private static final int RESULT_COUNT_LEN = 4;
+ private static final int RESULT_ENTRY_LEN = 8;
+ private static final int REGISTER_BODY_MIN_LEN = 17;
+
+ private final Queue<CompletableFuture<ByteBuf>> responseQueue = new ConcurrentLinkedQueue<>();
+ private final ConsensusSession session;
+ private final Runnable onEviction;
+
+ public VsrResponseHandler(ConsensusSession session, Runnable onEviction) {
+ this.session = session;
+ this.onEviction = onEviction;
+ }
+
+ public void enqueueRequest(CompletableFuture<ByteBuf> future) {
+ responseQueue.add(future);
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ failPendingRequests(new IggyConnectionException("Connection closed before a response arrived"));
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ failPendingRequests(cause);
+ ctx.close();
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+ CompletableFuture<ByteBuf> future = responseQueue.poll();
+ if (future == null) {
+ log.error(
+ "Received response on channel {} but no request was waiting!",
+ ctx.channel().id());
+ return;
+ }
+ ByteBuf body;
+ try {
+ body = decodeReply(msg);
+ } catch (RuntimeException error) {
+ future.completeExceptionally(error);
+ return;
+ }
+ future.complete(body);
+ }
Review Comment:
The handler polls `responseQueue` and returns before inspecting the consensus command. Evictions are session-level, one-way frames and can arrive while the connection is idle, notably from the stale-client verifier. Such a frame is currently discarded, so the session remains bound, the authentication generation is not advanced, and cached consumer-group assignments survive after the server has removed them. Consider inspecting and applying eviction effects before requiring a request future and failing any pending requests after resetting the session.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java:
##########
@@ -108,16 +146,19 @@ public AsyncTcpConnection(
.option(ChannelOption.SO_KEEPALIVE, true)
.remoteAddress(host, port);
+ // The VSR session (client id, fence epoch, request counter) is bound
+ // to one transport connection server-side; sharing it across channels
+ // would interleave request ids, so the pool holds a single channel.
this.channelPool = new FixedChannelPool(
bootstrap,
- new PoolChannelHandler(host, port, enableTls, sslContext),
+ new PoolChannelHandler(host, port, enableTls, sslContext, consensusSession, this::onSessionEvicted),
ChannelHealthChecker.ACTIVE,
FixedChannelPool.AcquireTimeoutAction.FAIL,
poolConfig.getAcquireTimeoutMillis(),
- poolConfig.getMaxConnections(),
+ 1,
poolConfig.getMaxPendingAcquires());
- log.info("Connection pool initialized with max connections: {}", poolConfig.getMaxConnections());
+ log.info("Connection pool initialized with a single VSR-pinned connection");
Review Comment:
This changes the pool from five configurable connections to exactly one while retaining `AcquireTimeoutAction.FAIL`. The same class defines the default acquire timeout as three seconds at line 595 and the transient retry budget as 30 seconds at line 88. Because the channel is held until the response finishes, one slow or transiently retried operation makes every concurrent request fail acquisition after three seconds. This is a direct regression for the async client's concurrent callers and would also starve any heartbeat task. Consider preserving session pinning without a short pending-acquire deadline, multiplexing requests on the pinned channel, or using a separate VSR session per pool channel.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrFrameDecoder.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp.vsr;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.DecoderException;
+
+import java.util.List;
+
+/**
+ * Decoder for VSR response frames: a 256-byte consensus header whose total
+ * frame size (header included) sits at byte offset 48, followed by an
+ * optional body. There is no length delimiter outside the header.
+ */
+public class VsrFrameDecoder extends ByteToMessageDecoder {
+
+ /** Matches the server's default {@code max_message_size} (64 MB). */
+ static final long MAX_FRAME_SIZE = 64L * 1024 * 1024;
+
+ @Override
+ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
Review Comment:
The decoder treats the server's 64 MiB default as a protocol maximum, while `core/server-ng/config.toml:1000` exposes `message_bus.max_message_size` as a deployment setting rather than a fixed wire limit. A server configured for larger batches can therefore send a valid poll or metadata response that this client rejects as a desynchronized connection. The previous Java decoder had no such fixed ceiling. Consider making the client limit configurable and ensuring that it agrees with the target deployment instead of hard-coding the server default.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ClientRoutingState.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp;
+
+import org.apache.iggy.identifier.ConsumerId;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Per-client cache of the routing facts needed to resolve partitioning and
+ * consumer-group polling client-side: topic partition counts, balanced
+ * round-robin cursors, and consumer-group assignments. Keys are
+ * {@code stream|topic[|group]} strings built from the request identifiers,
+ * matching the Rust SDK's {@code ConsumerGroupClientState}.
+ *
+ * <p>Partition counts and balanced cursors survive reconnects; group
+ * assignments are bound to the server-side VSR session (the member is keyed
+ * by the connection's client id) and must be cleared whenever that session is
+ * reset or the client moves to another node. Partition counts carry their
+ * fetch timestamp so callers can refresh them past a staleness budget — a
+ * count cached forever would keep hashing keys with the wrong modulus after
+ * a partition-count change (the Rust SDK still has that gap).
+ */
+final class ClientRoutingState {
+
+ private final Map<String, CachedPartitionCount> partitionCounts = new ConcurrentHashMap<>();
+ private final Map<String, AtomicInteger> balancedCursors = new ConcurrentHashMap<>();
+ private final Map<String, GroupAssignment> assignments = new ConcurrentHashMap<>();
+
+ static String topicKey(StreamId streamId, TopicId topicId) {
+ return streamId + "|" + topicId;
+ }
+
+ static String groupKey(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ return streamId + "|" + topicId + "|" + groupId;
+ }
Review Comment:
The `streamId + "|" + topicId` and `streamId + "|" + topicId + "|" + groupId` concatenations implicitly convert each identifier by calling the implementation at `identifier/Identifier.java:50-56`. That method returns only the name or numeric value and omits the identifier kind, while names can also contain the `|` delimiter. For example, numeric stream/topic IDs `1` and `2` produce the same key as named identifiers `"1"` and `"2"`, even though they can resolve to different resources; delimiter-containing names create further collisions. The cache can then reuse another topic's partition count or another group's assignment, causing sends to be routed to an invalid partition and polls to use an unrelated assignment. Consider keying these maps with records containing each identifier's kind and value rather than a formatted string.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrNamespace.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp.vsr;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+import org.apache.iggy.exception.IggyServerException;
+
+/**
+ * Partition-plane namespace packing and payload peeking, mirroring
+ * {@code core/binary_protocol/src/namespace.rs} and the request inspection in
+ * {@code core/sdk/src/vsr.rs}.
+ *
+ * <p>The packed namespace routes a request to its partition's consensus
+ * group: stream id in bits 32..43, topic id in bits 20..31, partition id in
+ * bits 0..19. Control-plane requests (Register/Logout) use the metadata
+ * sentinel {@code 1 << 63}; non-replicated and metadata ops use zero.
+ */
+final class VsrNamespace {
+
+ static final long METADATA_CONSENSUS_NAMESPACE = 1L << 63;
+
+ static final int MAX_STREAMS = 4096;
+ static final int MAX_TOPICS = 4096;
+ static final int MAX_PARTITIONS = 1_000_000;
+
+ static final int STREAM_SHIFT = 32;
+ static final int TOPIC_SHIFT = 20;
+
+ private static final int SEND_MESSAGES_CODE = 101;
+ private static final int STORE_CONSUMER_OFFSET_CODE = 121;
+ private static final int DELETE_CONSUMER_OFFSET_CODE = 122;
+ private static final int STORE_CONSUMER_OFFSET_2_CODE = 123;
+ private static final int DELETE_CONSUMER_OFFSET_2_CODE = 124;
+
+ private static final int IDENTIFIER_KIND_NUMERIC = 1;
+ private static final int PARTITIONING_KIND_PARTITION_ID = 2;
+
+ private VsrNamespace() {}
+
+ static long namespaceFor(int operation, int commandCode, ByteBuf payload) {
+ if (operation == VsrOperation.REGISTER || operation == VsrOperation.LOGOUT) {
+ return METADATA_CONSENSUS_NAMESPACE;
+ }
+ if (operation == VsrOperation.NON_REPLICATED || VsrOperation.isMetadata(operation)) {
+ return 0;
+ }
+ return switch (commandCode) {
+ case SEND_MESSAGES_CODE -> fromSendMessages(payload);
+ case STORE_CONSUMER_OFFSET_CODE,
+ DELETE_CONSUMER_OFFSET_CODE,
+ STORE_CONSUMER_OFFSET_2_CODE,
+ DELETE_CONSUMER_OFFSET_2_CODE -> fromConsumerOffset(payload);
+ default -> throw IggyServerException.fromTcpResponse(VsrHeaders.ERROR_FEATURE_UNAVAILABLE, new byte[0]);
Review Comment:
`VsrOperation.java:105` maps command 503 to `DELETE_SEGMENTS`, but that operation is deliberately neither metadata nor partition-classified. It therefore reaches this switch and falls into `FeatureUnavailable` for every request, so `sendBinaryRequest(503, ...)` can never reach server-ng. Consider adding the Rust SDK's equivalent case, which decodes the delete-segments request and packs its stream, topic, and partition into the namespace.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java:
##########
@@ -177,13 +180,45 @@ public static ConsumerGroup readConsumerGroup(ByteBuf response) {
return new ConsumerGroup(groupId, name, partitionsCount, membersCount);
}
+ public static ConsumerGroupAssignment readConsumerGroupAssignment(ByteBuf response) {
+ // The generation is a monotonic rebalance counter compared only for
+ // equality, so reading the u64 as a signed long is safe.
+ var generation = response.readLongLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ List<Long> partitions = new ArrayList<>(toInt(partitionsCount));
+ for (long i = 0; i < partitionsCount; i++) {
+ partitions.add(response.readUnsignedIntLE());
+ }
+ return new ConsumerGroupAssignment(generation, partitions);
Review Comment:
ALso applies to BytesDeserializer.java:202-219
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(java): migrate the TCP client to the VSR wire protocol [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3841:
URL: https://github.com/apache/iggy/pull/3841#discussion_r3740609144
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java:
##########
@@ -193,100 +234,225 @@ public CompletableFuture<ByteBuf> send(CommandCode commandCode, ByteBuf payload)
}
public CompletableFuture<ByteBuf> send(int commandCode, ByteBuf payload) {
+ if (isLoginCode(commandCode) && authenticated) {
+ return logoutThenLogin(commandCode, payload);
+ }
captureLoginPayloadIfNeeded(commandCode, payload);
CompletableFuture<ByteBuf> responseFuture = new CompletableFuture<>();
CompletableFuture<ByteBuf> callerFuture = new CompletableFuture<>();
channelPool.acquire().addListener((FutureListener<Channel>) f -> {
if (!f.isSuccess()) {
payload.release();
+ notifyConnectionFailure(f.cause());
callerFuture.completeExceptionally(mapAcquireException(f.cause()));
return;
}
+ dispatchOnChannel(f.getNow(), commandCode, payload, responseFuture, callerFuture);
+ });
- Channel channel = f.getNow();
- boolean isLoginCommand = (commandCode == CommandCode.User.LOGIN.getValue()
- || commandCode == CommandCode.PersonalAccessToken.LOGIN.getValue());
- boolean requiresAuth = !isLoginCommand
- && commandCode != CommandCode.System.PING.getValue()
- && commandCode != CommandCode.System.GET_STATS.getValue();
+ return callerFuture;
+ }
- responseFuture.whenComplete((response, error) -> {
- try {
- handlePostResponse(channel, commandCode, isLoginCommand, error);
- } catch (RuntimeException bookkeepingError) {
- log.error("Post-response bookkeeping failed: {}", bookkeepingError.getMessage());
- }
- if (error != null) {
- callerFuture.completeExceptionally(error);
- } else {
- callerFuture.complete(response);
- }
- });
+ private void dispatchOnChannel(
+ Channel channel,
+ int commandCode,
+ ByteBuf payload,
+ CompletableFuture<ByteBuf> responseFuture,
+ CompletableFuture<ByteBuf> callerFuture) {
+ boolean isLoginCommand = isLoginCode(commandCode);
+ boolean requiresAuth = !isLoginCommand && requiresAuthentication(commandCode);
- CompletableFuture<Void> authStep;
- if (!requiresAuth) {
- authStep = CompletableFuture.completedFuture(null);
- } else if (!authenticated) {
+ responseFuture.whenComplete((response, error) -> {
+ try {
+ handlePostResponse(channel, commandCode, isLoginCommand, error);
+ } catch (RuntimeException bookkeepingError) {
+ log.error("Post-response bookkeeping failed: {}", bookkeepingError.getMessage());
+ }
+ if (error != null) {
+ callerFuture.completeExceptionally(error);
+ } else {
+ callerFuture.complete(response);
+ }
+ });
+
+ CompletableFuture<Void> authStep;
+ if (!requiresAuth) {
+ authStep = CompletableFuture.completedFuture(null);
+ } else if (!authenticated) {
+ payload.release();
+ responseFuture.completeExceptionally(new IggyNotConnectedException("Not authenticated, call login first"));
+ return;
+ } else {
+ ByteBuf loginPayloadCopy = getLoginPayloadCopy();
+ if (loginPayloadCopy == null) {
payload.release();
responseFuture.completeExceptionally(
new IggyNotConnectedException("Not authenticated, call login first"));
return;
- } else {
- ByteBuf loginPayloadCopy = getLoginPayloadCopy();
- if (loginPayloadCopy == null) {
- payload.release();
- responseFuture.completeExceptionally(
- new IggyNotConnectedException("Not authenticated, call login first"));
- return;
- }
- authStep = IggyAuthenticator.ensureAuthenticated(
- channel, loginPayloadCopy, loginCommandCode, authGeneration);
}
+ authStep = IggyAuthenticator.ensureAuthenticated(
+ channel, loginPayloadCopy, loginCommandCode, authGeneration, vsrEncoder);
+ }
- authStep.thenRun(() -> sendFrame(channel, payload, commandCode, responseFuture))
- .exceptionally(ex -> {
- responseFuture.completeExceptionally(ex);
- return null;
- });
+ authStep.whenComplete((ignored, authError) -> {
+ if (authError != null) {
+ payload.release();
+ responseFuture.completeExceptionally(authError);
+ return;
+ }
+ sendFrame(channel, payload, commandCode, responseFuture);
});
+ }
- return callerFuture;
+ /**
+ * A failed pool acquire on an established connection means the target
+ * node could not be (re)dialed; the listener lets the owning client run
+ * its redial strategy while the failed request surfaces to its caller.
+ */
+ private void notifyConnectionFailure(Throwable cause) {
+ try {
+ connectionFailureListener.accept(cause);
+ } catch (RuntimeException listenerError) {
+ log.warn("Connection failure listener threw: {}", listenerError.getMessage());
+ }
}
private static Throwable mapAcquireException(Throwable cause) {
if (cause instanceof IllegalStateException) {
return new IggyNotConnectedException("Connection pool is closed");
}
+ if (cause instanceof TimeoutException) {
+ return new IggyTimeoutException("Timed out acquiring a connection from the pool", cause);
+ }
return cause;
}
+ /**
+ * A Register on an already-bound VSR connection is answered with a replay
+ * of the original register reply, while the client has re-armed a fresh
+ * identity; its reset request counter would then collide with the
+ * server's dedup table and mutations would be silently swallowed. Unbind
+ * first, then login fresh.
+ */
+ private CompletableFuture<ByteBuf> logoutThenLogin(int commandCode, ByteBuf payload) {
+ return send(CommandCode.User.LOGOUT.getValue(), Unpooled.EMPTY_BUFFER)
+ .handle((logoutResponse, logoutError) -> {
+ if (logoutResponse != null) {
+ logoutResponse.release();
+ }
+ return null;
+ })
+ .thenCompose(ignored -> send(commandCode, payload));
+ }
+
+ private static boolean isLoginCode(int commandCode) {
+ return commandCode == CommandCode.User.LOGIN.getValue()
+ || commandCode == CommandCode.PersonalAccessToken.LOGIN.getValue();
+ }
+
+ /**
+ * Ping and cluster metadata are the only sessionless bootstrap commands.
+ * Cluster metadata must be available before Register so a VSR client can
+ * select the leader; every other non-login command requires a bound
+ * session.
+ */
+ private static boolean requiresAuthentication(int commandCode) {
+ return !isAllowedBeforeAuthentication(commandCode);
+ }
+
+ private static boolean isAllowedBeforeAuthentication(int commandCode) {
+ return commandCode == CommandCode.System.PING.getValue()
+ || commandCode == CommandCode.System.GET_CLUSTER_METADATA.getValue();
+ }
+
private void sendFrame(
Channel channel, ByteBuf payload, int commandCode, CompletableFuture<ByteBuf> responseFuture) {
try {
- IggyResponseHandler handler = channel.pipeline().get(IggyResponseHandler.class);
+ VsrResponseHandler handler = channel.pipeline().get(VsrResponseHandler.class);
if (handler == null) {
- throw new IggyClientException("Channel missing IggyResponseHandler");
+ throw new IggyClientException("Channel missing VsrResponseHandler");
}
- handler.enqueueRequest(responseFuture);
- ByteBuf frame = IggyFrameEncoder.encode(channel.alloc(), commandCode, payload);
-
- channel.writeAndFlush(frame).addListener((ChannelFutureListener) future -> {
- if (!future.isSuccess()) {
- log.error("Failed to send frame: {}", future.cause().getMessage());
- responseFuture.completeExceptionally(future.cause());
- } else {
- log.trace("Frame sent successfully to {}", channel.remoteAddress());
- }
- });
+ ByteBuf frame = vsrEncoder.encode(channel.alloc(), commandCode, payload);
+ long nowNanos = System.nanoTime();
+ long deadlineNanos = nowNanos + TRANSIENT_RETRY_BUDGET.toNanos();
+ long notAcceptedDeadlineNanos =
+ isLoginCode(commandCode) ? deadlineNanos : nowNanos + NOT_ACCEPTED_RETRY_BUDGET.toNanos();
+ writeVsrFrame(channel, handler, frame, responseFuture, deadlineNanos, notAcceptedDeadlineNanos);
} catch (RuntimeException e) {
responseFuture.completeExceptionally(e);
} finally {
payload.release();
}
}
Review Comment:
This was resolved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org