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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
