You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/13 20:34:49 UTC
(iggy) 01/06: fix(sdk): preserve routed admission and history freshness
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 c98e26680bb2e8de0d6dfe650c9d130eee963692
Author: Hubert Gruszecki <h....@gmail.com>
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);