This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6dbe589b1bf4a03b51a53be38f79265679a60120 Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 22:33:14 2026 +0200 fix(sdk): preserve routed admission and history freshness Queued Go requests consumed IDs before endpoint admission, and stale Java/Node routes could survive newer metadata. C# also reported an unknown outcome for a generation change before any data send. Stamp after admission, expire stale routes without replacing captured history, classify unsent resets as not accepted, and require explicit replica IDs in Java/C# JSON metadata. --- foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs | 2 +- .../TcpMessageStream.PollRouting.cs | 5 +- .../Iggy_SDK_Tests/MapperTests/BinaryMapper.cs | 32 +++++++++++++ .../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 29 ++++++++++++ foreign/go/client/tcp/tcp_poll_routing.go | 17 +++---- foreign/go/client/tcp/tcp_poll_routing_test.go | 48 ++++++++++++++++++++ .../apache/iggy/client/async/tcp/PollRouter.java | 4 +- .../java/org/apache/iggy/cluster/ClusterNode.java | 6 +++ .../AsyncIggyTcpClientTransientFailoverTest.java | 12 +++-- .../iggy/client/async/tcp/PollRouterTest.java | 50 +++++++++++++++++--- .../client/blocking/http/ObjectMapperTest.java | 19 ++++++++ foreign/node/src/client/client.socket.test.ts | 53 ++++++++++++++++++++-- foreign/node/src/client/client.socket.ts | 2 +- 13 files changed, 250 insertions(+), 29 deletions(-) diff --git a/foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs b/foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs index 2c91b3249..4db860efa 100644 --- a/foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs +++ b/foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs @@ -28,7 +28,7 @@ public class ClusterNode public required string Name { get; set; } /// <summary>Configured replica identity, independent of its position in the cluster roster.</summary> - public byte ReplicaId { get; set; } + public required byte ReplicaId { get; set; } /// <summary> /// Node IP address diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs index 21b7d2808..a4c3a9f59 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs @@ -186,7 +186,7 @@ public sealed partial class TcpMessageStream captured ??= route.Route.Context; if (generation is not null && generation != route.Generation) { - throw new VsrRequestOutcomeUnknownException(new NotConnectedException()); + throw PollNotAccepted(); } generation ??= route.Generation; frame ??= _consensusSession.Resolve(VsrOperations.ForCode(code)); @@ -213,7 +213,8 @@ public sealed partial class TcpMessageStream VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.NOT_PRIMARY) { RemovePartitionRoute(target.Key); - if (Environment.TickCount64 + PollRoutingRetryIntervalMs >= deadline) + if ((generation is not null && generation != _consensusSession.Generation) + || Environment.TickCount64 + PollRoutingRetryIntervalMs >= deadline) { throw; } diff --git a/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs b/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs index f9f031d8f..9def08cbf 100644 --- a/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs +++ b/foreign/csharp/Iggy_SDK_Tests/MapperTests/BinaryMapper.cs @@ -18,6 +18,8 @@ using System.Buffers.Binary; using System.Security.Cryptography; using System.Text; +using System.Text.Json; +using System.Text.Json.Serialization; using Apache.Iggy.Contracts; using Apache.Iggy.Contracts.Auth; using Apache.Iggy.Encryption; @@ -39,6 +41,36 @@ namespace Apache.Iggy.Tests.MapperTests; public sealed class BinaryMapper { + [Theory] + [InlineData(null)] + [InlineData("null")] + [InlineData("0")] + [InlineData("7")] + [InlineData("255")] + public void given_cluster_json_when_reading_replica_identity_should_require_an_explicit_id(string? replica) + { + var identity = replica is null ? "" : $",\"replica_id\":{replica}"; + var json = """ + {"name":"node","ip":"127.0.0.1","endpoints":{"tcp":8090,"quic":8080,"http":3000,"web_socket":8092}, + "role":"leader","status":"healthy" + """ + identity + "}"; + var options = new JsonSerializerOptions + { + PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower, + Converters = { new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower) } + }; + if (replica is null or "null") + { + Assert.Throws<JsonException>(() => JsonSerializer.Deserialize<ClusterNode>(json, options)); + } + else + { + var node = JsonSerializer.Deserialize<ClusterNode>(json, options); + Assert.NotNull(node); + Assert.Equal(byte.Parse(replica), node.ReplicaId); + } + } + [Fact] public void MapPersonalAccessTokens_ReturnsValidPersonalAccessTokenResponse() { diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs index ddac29e64..e933c4a18 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs @@ -505,6 +505,35 @@ public sealed class PrimaryPollingTests Assert.Equal(1, cluster.Coordinator.Registrations); } + [Fact] + public async Task given_unsent_poll_when_its_session_is_replaced_should_report_not_accepted() + { + using var cluster = new PollCluster(); + var binding = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + cluster.OnPrimaryAttachment = request => + { + binding.SetResult(); + release.Task.GetAwaiter().GetResult(); + return BindingReply(request); + }; + using var client = await cluster.ConnectAsync(); + var pending = PollAsync(client); + try + { + await binding.Task.WaitAsync(TestContext.Current.CancellationToken); + await client.LogoutUserAsync(TestContext.Current.CancellationToken); + await client.LoginUserAsync("other", "secret", TestContext.Current.CancellationToken); + var error = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() => pending); + Assert.Equal(VsrError.TRANSIENT_NOT_ACCEPTED, error.StatusCode); + Assert.All(cluster.Primaries, primary => Assert.Equal(0, primary.Requests(CommandCodes.POLL_MESSAGES_CODE))); + } + finally + { + release.TrySetResult(); + } + } + [Fact] public async Task given_new_coordinator_session_when_polling_should_retire_old_data_connections() { diff --git a/foreign/go/client/tcp/tcp_poll_routing.go b/foreign/go/client/tcp/tcp_poll_routing.go index ec833d1cf..a84509081 100644 --- a/foreign/go/client/tcp/tcp_poll_routing.go +++ b/foreign/go/client/tcp/tcp_poll_routing.go @@ -48,6 +48,7 @@ type singlePollExchange struct{} type capturedPartitionContext struct{} type pollExchangeState struct { + stamped bool written bool reusable bool replay bool @@ -212,7 +213,7 @@ func (c *IggyTcpClient) sendRouted(caller context.Context, code command.Code, qu if requestParent != nil && !c.matchesPollParent(*requestParent) { return nil, ierror.ErrUnauthenticated } - state := pollExchangeState{replay: isPartitionMutation(code)} + state := pollExchangeState{stamped: stamped, replay: isPartitionMutation(code)} route, err := c.consumerRoute(ctx, query) if err == nil { if requestParent != nil && (requestParent.client != route.parent.client || requestParent.session != route.parent.session) { @@ -231,19 +232,17 @@ func (c *IggyTcpClient) sendRouted(caller context.Context, code command.Code, qu if snapshot == nil || snapshot.parent.client != route.parent.client || snapshot.parent.session != route.parent.session { return nil, ierror.ErrUnauthenticated } - if err := vsr.StampRequestHeader(snapshot.session, uint32(code), frame); err != nil { - return nil, err - } - stamped = true parent := route.parent requestParent = &parent } vsr.StampPartitionContext((*[vsr.HeaderSize]byte)(frame[:vsr.HeaderSize]), route.context) if c.clustered.Load() { response, err = c.pollOnRoute(ctx, code, key, frame, route, &state) + stamped = state.stamped } else { state.parent = requestParent - response, _, _, err = c.attempt(context.WithValue(ctx, singlePollExchange{}, &state), uint32(code), frame, true, time.Now(), time.Now().Add(responseReadTimeout)) + exchangeCtx := context.WithValue(ctx, capturedPartitionContext{}, route.context) + response, stamped, _, err = c.attempt(context.WithValue(exchangeCtx, singlePollExchange{}, &state), uint32(code), frame, stamped, time.Now(), time.Now().Add(responseReadTimeout)) if state.written && !state.reusable && err != nil { err = ierror.ErrTransientNotCommitted } @@ -404,8 +403,10 @@ func (c *IggyTcpClient) pollOnRoute(ctx context.Context, code command.Code, key if !c.pollParentCurrent(route.parent) { return nil, ierror.ErrTransientNotAccepted } - response, _, _, err := slot.client.attempt(context.WithValue(exchangeCtx, singlePollExchange{}, state), - uint32(code), frame, true, time.Now(), time.Now().Add(responseReadTimeout)) + exchangeCtx = context.WithValue(exchangeCtx, capturedPartitionContext{}, route.context) + response, stamped, _, err := slot.client.attempt(context.WithValue(exchangeCtx, singlePollExchange{}, state), + uint32(code), frame, state.stamped, time.Now(), time.Now().Add(responseReadTimeout)) + state.stamped = stamped err = slot.exchangeError(ctx, *state, err) if errors.Is(err, ierror.ErrTransientNotAccepted) { slot.attached = false diff --git a/foreign/go/client/tcp/tcp_poll_routing_test.go b/foreign/go/client/tcp/tcp_poll_routing_test.go index 77a8dc115..3a62be934 100644 --- a/foreign/go/client/tcp/tcp_poll_routing_test.go +++ b/foreign/go/client/tcp/tcp_poll_routing_test.go @@ -1500,3 +1500,51 @@ func TestSingleNodeQueuedPartitionRequestCannotUseAReplacementSession(t *testing assert.Equal(t, 1, requestCount(server.recorded(), command.PollMessagesCode)) }) } + +func TestQueuedPartitionRequestAllocatesIdentityAfterConnectionAdmission(t *testing.T) { + for _, clustered := range []bool{false, true} { + t.Run(fmt.Sprint(clustered), func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + client, serverConn := newPipeClient(t) + servePartitionOperations(t, serverConn, func(_ int, _ request) []byte { + return replyFrame(vsr.OperationNonReplicated, emptyBatchBody(0)) + }) + stream, topic, group := groupConsumer(t) + consumer := iggcon.NewSingleConsumer(group.Id) + poll := func() error { + _, err := client.PollMessages(context.Background(), stream, topic, consumer, + iggcon.NextPollingStrategy(), 1, false, nil) + return err + } + require.NoError(t, poll()) + gate := client.exchangeGate + if clustered { + data, peer := newPipeClient(t) + serve(peer, func(_ int, _ request) []byte { + return replyFrame(vsr.OperationNonReplicated, emptyBatchBody(0)) + }) + data.session = client.session.Fork() + lifetime, cancel := context.WithCancel(context.Background()) + defer cancel() + gate = make(chan struct{}, 1) + for _, route := range client.polls.routes { + client.polls.connections = map[string]*pollConnection{route.endpoint: { + gate: gate, ctx: lifetime, cancel: cancel, conn: data.conn, + client: data, parent: route.parent, attached: true, + }} + } + client.clustered.Store(true) + } + gate <- struct{}{} + before := client.session.CurrentRequestID() + completed := make(chan error, 1) + go func() { completed <- poll() }() + synctest.Wait() + assert.Equal(t, before, client.session.CurrentRequestID(), "a queued request must not age the shared replay window") + <-gate + require.NoError(t, <-completed) + assert.Equal(t, before+1, client.session.CurrentRequestID()) + }) + }) + } +} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java index 967762b90..06bbf8708 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java @@ -410,7 +410,9 @@ final class PollRouter { private synchronized boolean routeIsCurrent(Route route) { AsyncTcpConnection parent = coordinator.get(); - return route.parent == parent && route.generation == parent.sessionGeneration(); + return route.parent == parent + && route.generation == parent.sessionGeneration() + && Long.compareUnsigned(route.attachment.watermark, parent.metadataWatermark()) >= 0; } private CompletableFuture<ByteBuf> enqueue(Route route, Poll poll) { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java index bc4855a36..67c20f865 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java @@ -19,6 +19,10 @@ package org.apache.iggy.cluster; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonSetter; +import com.fasterxml.jackson.annotation.Nulls; + /** * A single node in the cluster roster. * @@ -30,7 +34,9 @@ package org.apache.iggy.cluster; * @param status the node status */ public record ClusterNode( + @JsonProperty(value = "replica_id", required = true) @JsonSetter(nulls = Nulls.FAIL) int replicaId, + String name, String ip, TransportEndpoints endpoints, diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java index 13f509dd1..1dd56af20 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java @@ -852,7 +852,8 @@ class AsyncIggyTcpClientTransientFailoverTest { if (request.is(GET_PARTITION_ROUTE_CODE, OPERATION_NON_REPLICATED)) { routes.incrementAndGet(); return Response.success( - OPERATION_NON_REPLICATED, pollRoute(request, 1, 1, primarySocket.getLocalPort())); + OPERATION_NON_REPLICATED, + pollRoute(request, 1, Math.max(1, request.metadataOp()), primarySocket.getLocalPort())); } if (request.operation() == OPERATION_CREATE_STREAM) { return Response.committed( @@ -896,7 +897,7 @@ class AsyncIggyTcpClientTransientFailoverTest { assertThat(attachments).containsExactly(1L, 11L); assertThat(pollWatermarks) .containsExactlyElementsOf(holdAttachment ? List.of(11L) : List.of(1L, 11L)); - assertThat(routes).hasValue(1); + assertThat(routes).hasValue(2); } finally { release.complete(null); client.close().get(5, TimeUnit.SECONDS); @@ -1052,7 +1053,7 @@ class AsyncIggyTcpClientTransientFailoverTest { pollRoute( request, 1, - 1, + Math.max(1, request.metadataOp()), partition(request) == 0 ? firstSocket.getLocalPort() : secondSocket.getLocalPort())); @@ -1109,7 +1110,7 @@ class AsyncIggyTcpClientTransientFailoverTest { poll(client, Optional.of(0L), true).get(5, TimeUnit.SECONDS); poll(client, Optional.of(0L), false).get(5, TimeUnit.SECONDS); assertThat(attachedWatermarks).containsExactly(1L, 1L, 11L); - assertThat(routes).hasValue(2); + assertThat(routes).hasValue(3); assertThat(firstPolls).hasValue(5); assertThat(ordinaryPolls).hasValue(0); assertThat(logins).hasValue(1); @@ -1149,7 +1150,8 @@ class AsyncIggyTcpClientTransientFailoverTest { } if (request.is(GET_PARTITION_ROUTE_CODE, OPERATION_NON_REPLICATED)) { return Response.success( - OPERATION_NON_REPLICATED, pollRoute(request, 1, 1, primarySocket.getLocalPort())); + OPERATION_NON_REPLICATED, + pollRoute(request, 1, Math.max(1, request.metadataOp()), primarySocket.getLocalPort())); } if (request.operation() == UPDATE_USER_OPERATION || request.operation() == CHANGE_PASSWORD_OPERATION) { diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/PollRouterTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/PollRouterTest.java index ccd9feb48..bc90567c0 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/PollRouterTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/PollRouterTest.java @@ -62,8 +62,8 @@ class PollRouterTest { private static final int PARTITION_KEY_BYTES = 16; private static final PartitionContext ROUTE_CONTEXT = new PartitionContext(BigInteger.valueOf(INCARNATION), BigInteger.ZERO, BigInteger.valueOf(ROUTE_WATERMARK)); - // Attachment validation reads the watermark, then the send checks it again. - private static final int SEND_CHECK_WATERMARK_READ = 2; + // Fire after route freshness checks, at the final attachment check before send. + private static final int SEND_CHECK_WATERMARK_READ = 6; private static final Duration TIMEOUT = Duration.ofSeconds(5); private static final Duration NEVER = Duration.ofDays(1); @@ -109,6 +109,33 @@ class PollRouterTest { } } + @Test + void shouldRefreshCachedHistoryBeforeANewPollAfterMetadataAdvances() throws Exception { + IoEventLoopGroup group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + ManualTimers timers = new ManualTimers(); + try { + Coordinator coordinator = new Coordinator(group, timers); + Primary primary = new Primary(group); + primary.holdAfterFirst = false; + PollRouter router = new PollRouter(() -> coordinator, endpoint -> primary); + router.poll(pollPayload(), Optional.empty()) + .get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS) + .release(); + coordinator.watermark = ROUTE_WATERMARK + 1; + coordinator.context = new PartitionContext( + BigInteger.valueOf(INCARNATION + 1), BigInteger.ZERO, BigInteger.valueOf(coordinator.watermark)); + router.poll(pollPayload(), Optional.empty()) + .get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS) + .release(); + assertThat(primary.sentContexts).containsExactly(ROUTE_CONTEXT, coordinator.context); + assertThat(coordinator.routes).isEqualTo(2); + router.clearSession().get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + } finally { + timers.shutdownGracefully(0, 1, TimeUnit.SECONDS).get(5, TimeUnit.SECONDS); + group.shutdownGracefully(0, 1, TimeUnit.SECONDS).get(5, TimeUnit.SECONDS); + } + } + private static ByteBuf pollPayload() { // The router keys routes by the bytes before the poll parameters and reads nothing else. return Unpooled.buffer() @@ -180,6 +207,9 @@ class PollRouterTest { private static final class Coordinator extends OfflineConnection { private final EventLoop timers; + private long watermark; + private PartitionContext context = ROUTE_CONTEXT; + private int routes; private final AtomicInteger watermarkReadsLeft = new AtomicInteger(); private volatile Runnable watermarkHook = () -> {}; @@ -209,7 +239,7 @@ class PollRouterTest { if (watermarkReadsLeft.decrementAndGet() == 0) { watermarkHook.run(); } - return super.metadataWatermark(); + return watermark; } @Override @@ -231,6 +261,7 @@ class PollRouterTest { PRIMARY_PORT + 1, PRIMARY_PORT, PRIMARY_PORT + 1)); } assertThat(commandCode).isEqualTo(CommandCode.Messages.GET_PARTITION_ROUTE.getValue()); + routes++; Request routing = new Request( OPERATION_NON_REPLICATED, GET_PARTITION_ROUTE_CODE, @@ -242,8 +273,12 @@ class PollRouterTest { 0, new byte[0], 0); - return CompletableFuture.completedFuture( - pollRoute(routing, SESSION, ROUTE_WATERMARK, PRIMARY_PORT, INCARNATION)); + return CompletableFuture.completedFuture(pollRoute( + routing, + SESSION, + context.metadataOp().longValue(), + PRIMARY_PORT, + context.incarnation().longValue())); } } @@ -251,6 +286,7 @@ class PollRouterTest { private static final class Primary extends OfflineConnection { private final List<PartitionContext> sentContexts = new CopyOnWriteArrayList<>(); private final CompletableFuture<ByteBuf> unanswered = new CompletableFuture<>(); + private boolean holdAfterFirst = true; Primary(IoEventLoopGroup group) { super(group); @@ -269,7 +305,9 @@ class PollRouterTest { return CompletableFuture.completedFuture(Unpooled.EMPTY_BUFFER); } sentContexts.add(state.context()); - return sentContexts.size() == 1 ? CompletableFuture.completedFuture(emptyPoll(0)) : unanswered; + return !holdAfterFirst || sentContexts.size() == 1 + ? CompletableFuture.completedFuture(emptyPoll(0)) + : unanswered; } @Override diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/ObjectMapperTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/ObjectMapperTest.java index 3b3c70442..3398784ce 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/ObjectMapperTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/ObjectMapperTest.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.blocking.http; +import org.apache.iggy.cluster.ClusterNode; import org.apache.iggy.message.Message; import org.apache.iggy.message.PolledMessages; import org.apache.iggy.partition.PartitionContext; @@ -35,11 +36,29 @@ import java.nio.charset.StandardCharsets; import java.util.Base64; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class ObjectMapperTest { private final ObjectMapper objectMapper = ObjectMapperFactory.getInstance(); + @Test + void shouldRequireAnExplicitReplicaIdInClusterJson() { + String node = """ + {"name":"node","ip":"127.0.0.1","endpoints":{"tcp":8090,"quic":8080,"http":3000,"websocket":8092}, + "role":"leader","status":"healthy"%s} + """; + for (String missing : new String[] {"", ",\"replica_id\":null"}) { + assertThatThrownBy(() -> objectMapper.readValue(node.formatted(missing), ClusterNode.class)) + .isInstanceOf(tools.jackson.core.JacksonException.class) + .hasMessageContaining("replica_id"); + } + for (int replica : new int[] {0, 7, 255}) { + var decoded = objectMapper.readValue(node.formatted(",\"replica_id\":" + replica), ClusterNode.class); + assertThat(decoded.replicaId()).isEqualTo(replica); + } + } + @Nested class Deserialization { diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts index c75c31da5..f0b12bfb3 100644 --- a/foreign/node/src/client/client.socket.test.ts +++ b/foreign/node/src/client/client.socket.test.ts @@ -266,7 +266,7 @@ const singleNodeHandler = (port: number): FrameHandler => return; } if (code === COMMAND_CODE.GetPartitionRoute) { - socket.write(replyFrame(Operation.NonReplicated, routingBody(1n))); + socket.write(replyFrame(Operation.NonReplicated, routingBody(frame.readBigUInt64LE(REQUEST_OFFSET.minimumMetadataOp) || 1n))); return; } socket.write(replyFrame(operation)); @@ -365,7 +365,7 @@ const startPollCluster = async ( return; } if (code === COMMAND_CODE.GetPartitionRoute) { - socket.write(replyFrame(Operation.NonReplicated, routingBody(1n))); + socket.write(replyFrame(Operation.NonReplicated, routingBody(frame.readBigUInt64LE(REQUEST_OFFSET.minimumMetadataOp) || 1n))); return; } singleNodeHandler(coordinator.port)(frame, socket); @@ -545,6 +545,49 @@ describe('unified partition routing', () => { } }); + for (const command of [COMMAND_CODE.PollMessages, COMMAND_CODE.GetOffset]) { + it(`refreshes cached history after observing a newer metadata commit for ${command}`, async () => { + let watermark = 1n; + let incarnation = 7n; + const histories: bigint[] = []; + const cluster = await startPollCluster((frame, socket) => { + if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== command) + return false; + histories.push(stampedContext(frame).incarnation); + socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0))); + return true; + }, (frame, socket) => { + if (frame[REQUEST_OFFSET.operation] === Operation.UpdateUser) { + watermark = 19n; + incarnation = 8n; + const response = replyFrame(Operation.UpdateUser, metadataResult(0)); + response.writeBigUInt64LE(watermark, REPLY_OFFSET.commit); + socket.write(response); + return true; + } + if (frame.readUInt32LE(REQUEST_OFFSET.reserved) !== COMMAND_CODE.GetPartitionRoute) + return false; + socket.write(replyFrame(Operation.NonReplicated, routingBody(watermark, incarnation))); + return true; + }); + try { + const payload = command === COMMAND_CODE.PollMessages + ? groupPollPayload() : GET_OFFSET.serialize({ streamId: 1, topicId: 2, + consumer: { kind: ConsumerKind.Single, id: 1 }, partitionId: 3 }); + await cluster.client.sendCommand(command, payload); + await cluster.client.sendCommand(COMMAND_CODE.UpdateUser, Buffer.alloc(0)); + await cluster.client.sendCommand(command, payload); + assert.deepEqual(histories, [7n, 8n]); + assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 2); + const lookups = cluster.coordinator.frames.filter((frame) => + frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.GetPartitionRoute); + assert.equal(stampedContext(lookups[1]).metadataOp, watermark); + } finally { + await cluster.close(); + } + }); + } + it('shares one route and request counter across all five partition commands', async () => { const cluster = await startPollCluster((frame, socket) => { const operation = frame[REQUEST_OFFSET.operation]; @@ -685,7 +728,7 @@ describe('primary auto-commit polling', () => { const attachments = cluster.primary.frames.filter((frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession); assert.deepEqual(attachments.map((frame) => bindingIdentity(frame).readBigUInt64LE(24)), [1n, 15n]); - assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 2); + assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetPartitionRoute), 3); await cluster.client.sendCommand(LOGOUT.code, LOGOUT.serialize()); await cluster.client.sendCommand(POLL_MESSAGES.code, groupPollPayload()); const replacement = cluster.primary.frames.filter((frame) => @@ -3056,7 +3099,7 @@ describe('VSR client socket', () => { }); } - it('reuses standalone routes across unrelated metadata commits', async () => { + it('refreshes standalone routes after metadata commits', async () => { const server = await startVsrServer((frame, socket) => { if (frame.readUInt8(REQUEST_OFFSET.operation) === Operation.CreateStream) { const reply = replyFrame(Operation.CreateStream, metadataResult(0)); @@ -3077,7 +3120,7 @@ describe('VSR client socket', () => { await client.sendCommand(COMMAND_CODE.CreateStream, Buffer.alloc(0)); await client.sendCommand(POLL_MESSAGES.code, groupPollPayload(false)); await client.sendCommand(COMMAND_CODE.StoreOffset, storeOffsetPayload); - assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 2); + assert.equal(countCommand(server, COMMAND_CODE.GetPartitionRoute), 4); } finally { client.destroy(); await server.close(); diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts index fe6f86169..35d1b64c2 100644 --- a/foreign/node/src/client/client.socket.ts +++ b/foreign/node/src/client/client.socket.ts @@ -502,7 +502,7 @@ export class CommandResponseStream extends EventEmitter { private async _lookupRoute(partition: PartitionTarget, deadline: number): Promise<PartitionRoute> { const cached = this.partitionRoutes.get(partition.key); - if (cached) + if (cached && cached.context.metadataOp >= this.metadataWatermark) return cached; const response = await this._routingControl(partition, deadline); return deserializePartitionRoute(response.data);
