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