numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4263203905
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -666,17 +767,161 @@ fn assert_committed_metadata_agrees(sim: &Simulator,
live: &[usize], seed: u64)
compared
}
-/// The live replica whose metadata consensus is the current primary, i.e. the
-/// authoritative holder of the committed metadata log.
-fn metadata_leader(sim: &Simulator, live: &[usize]) -> Option<usize> {
- live.iter().copied().find(|&replica_idx| {
- sim.replicas[replica_idx].shards[0]
- .plane
- .metadata()
- .consensus
- .as_ref()
- .is_some_and(|consensus| consensus.is_primary() &&
consensus.status() == Status::Normal)
- })
+/// Require the partition primary to hold each owner that leader metadata
settled,
+/// and equally committed replicas to agree. Returns how many owners it
compared.
+///
+/// Metadata settles an owner only after a partition installed it, so a settled
+/// owner the primary lacks was lost or never installed. Metadata names no
owner
+/// for an unassigned partition, which the partition installs as an owner with
no
+/// client.
+fn assert_settled_owners_installed(
+ sim: &Simulator,
+ workload: &Workload,
+ leader: usize,
+ seed: u64,
+) -> usize {
+ let streams = sim.replicas[leader].shards[0]
+ .plane
+ .metadata()
+ .mux_stm
+ .streams();
+ let mut compared = 0;
+ for &ns in &workload.options.namespaces {
+ let Some(primary) = sim.primary_index(ns) else {
+ continue;
+ };
+ let settled: Vec<(u64, Option<ConsumerGroupOwner>)> =
streams.read(|inner| {
+ let Some(topic) = inner
+ .items
+ .get(ns.stream_id())
+ .and_then(|stream| stream.topics.get(ns.topic_id()))
+ else {
+ return Vec::new();
+ };
+ topic
+ .consumer_groups
+ .iter()
+ .filter_map(|(&group_id, group)| {
+ let assignment =
group.assignments.get(&ns.partition_id())?;
+ assignment
+ .pending
+ .is_none()
+ .then_some((group_id, assignment.owner))
+ })
+ .collect()
+ });
+ for (group_id, owner) in settled {
+ let installed = sim.installed_group_owner(usize::from(primary),
ns, group_id);
+ let agrees = owner.map_or_else(
+ || installed.is_none_or(ConsumerGroupOwner::is_unassigned),
+ |owner| installed == Some(owner),
+ );
+ assert!(
+ agrees,
+ "leader {leader} settled owner {owner:?} of consumer group
{group_id} on \
+ ns {ns:?}, but partition primary {primary} installed
{installed:?} \
+ (seed={seed:#x})",
+ );
+ let primary_state =
sim.partition_consensus_state(usize::from(primary), ns);
+ for replica in 0..sim.replica_count {
+ let index = usize::from(replica);
+ if sim.is_crashed(replica)
+ || sim.partition_incarnation(index, ns)
+ != sim.partition_incarnation(usize::from(primary), ns)
+ || sim
+ .partition_consensus_state(index, ns)
+ .map(|state| state.commit_min)
+ != primary_state.as_ref().map(|state| state.commit_min)
+ {
+ continue;
+ }
+ assert_eq!(
+ sim.installed_group_owner(index, ns, group_id),
+ installed,
+ "replica {replica} and primary {primary} installed
different owners at the same committed prefix \
+ (ns={ns:?}, group={group_id}, seed={seed:#x})",
+ );
+ }
+ compared += 1;
+ }
+ }
+ compared
+}
+
+fn assert_partition_offsets_agree(sim: &Simulator, workload: &Workload, live:
&[usize], seed: u64) {
+ for &namespace in &workload.options.namespaces {
+ let mut committed = BTreeMap::new();
+ for &replica in live {
+ let Some(partition) = sim.replicas[replica]
+ .partition_shard(namespace)
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ else {
+ continue;
+ };
+ let mut offsets = BTreeMap::new();
+ for kind in ConsumerKind::ALL {
+ for (consumer, offset) in
partition.retained_consumer_offsets(kind) {
+ offsets.insert((kind.as_code(), consumer), offset);
+ }
+ }
+ let prefix = (
+ partition.created_revision(),
+ partition.consensus().commit_min(),
+ );
+ if let Some((other, expected)) = committed.get(&prefix) {
+ assert_eq!(
+ &offsets, expected,
+ "replicas {replica} and {other} have different consumer
offsets at committed prefix {prefix:?} \
+ (ns={namespace:?}, seed={seed:#x})"
+ );
+ }
+ if workload.strict_outcome_oracle()
+ && workload.options.ack_quorum_ratio >= 1.0
Review Comment:
nit: gated on `ack_quorum_ratio >= 1.0` while the default is 0.5, so the
shadow offset compare is off in every default campaign. 16 strict seeds
(clients=1, ack=1.0) pass with no false positive.
Add one strict cell to CI.
##########
core/simulator/src/workload/state_checker.rs:
##########
@@ -265,6 +381,53 @@ impl StateChecker {
}
}
+/// Membership is metadata state; the partition's ordered install barrier is
+/// what fences a departed member's later offset writes.
+fn check_partition_owner(
+ owners: &mut BTreeMap<u64, ConsumerGroupOwner>,
+ header: &PrepareHeader,
+ body: &[u8],
+ seed: u64,
+) {
+ if header.operation == Operation::InstallConsumerGroupOwner {
+ let install = InstallConsumerGroupOwnerRequest::decode_from(body)
+ .expect("a committed owner installation decodes");
+ owners.insert(install.group_id, install.owner);
+ return;
+ }
+ if is_auto_commit_client(header.client) {
Review Comment:
nit: auto-commit stores skip the generation check entirely, and `owners`
ignores `RetireConsumerGroupOwners`, so a stale store after retirement would
pass.
Drop retired groups when that op applies and check the generation for
auto-commit stores too.
##########
core/sdk/src/vsr.rs:
##########
@@ -125,26 +126,22 @@ pub(crate) fn encode_request_header(
}
_ => {
let operation = operation_for_code(code);
- // NonReplicated ops (ping, reads) bypass server-side dedup --
- // `ClientTable` only tracks request_ids for replicated ops, and
- // the table accepts any id above the watermark with no
- // contiguity requirement (client_table.rs: "There is no
- // `RequestGap`"), so consuming the counter would not break the
- // next metadata op. Read the current id without advancing
- // because the server ignores it for NonReplicated and burning
- // ids for requests the table never sees buys nothing.
+ // NonReplicated ops (ping, reads) bypass server-side dedup, and
+ // `ClientTable` admits any id above its watermark
+ // (client_table.rs: "There is no `RequestGap`"), so a read can
+ // spend an id. A bound session spends one per read, so a reply
+ // names exactly one request on a link that matches by id.
//
- // They are also sessionless on the server (routed by transport
- // id; protected codes are auth-gated server-side), so send with
- // session 0 when unregistered -- preserving the legacy "ping
- // works without auth" contract. Replicated ops still require a
- // bound session for dedup and fail fast here.
+ // Unbound, they are sessionless on the server (routed by
+ // transport id; protected codes are auth-gated server-side), so
+ // they carry session 0, which keeps the legacy "ping works
+ // without auth" contract. Replicated ops still require a bound
+ // session for dedup and fail fast here.
if operation == Operation::NonReplicated {
- (
- operation,
- session.current_request_id(),
- session.session().unwrap_or(0),
- )
+ match session.session() {
+ Some(session_id) => (operation,
session.next_request_id()?, session_id),
Review Comment:
nit: bound reads now spend a shared request id, here and in the Go/Java/Node
session counters; master did not. The 128-id committed window
(`client_table.rs:493-512`) is documented in terms of writes, so polls and
pings dilute it: a replayed write behind a busy poller can get `RequestTooOld`.
That fails safe (unknown outcome), but it isn't what the doc says.
Either use a separate counter for reads or update the window doc.
##########
foreign/go/client/tcp/tcp_poll_routing.go:
##########
@@ -200,33 +198,90 @@ func (c *IggyTcpClient) sendRouted(caller
context.Context, code command.Code, ke
c.polls.mu.Unlock()
if heartbeatDue {
if _, _, err := c.sendPollRequest(ctx,
uint32(command.PingCode), nil); err != nil {
Review Comment:
nit: the 5 s heartbeat PING runs synchronously in the caller's send/poll
path (+1 RTT on that request). There are also `string(query)` copies at :191
and :421, and the router mutex is taken 3x per op.
Move the heartbeat to a background ticker and reuse the key.
##########
foreign/go/client/tcp/tcp_poll_routing.go:
##########
@@ -200,33 +198,90 @@ func (c *IggyTcpClient) sendRouted(caller
context.Context, code command.Code, ke
c.polls.mu.Unlock()
if heartbeatDue {
if _, _, err := c.sendPollRequest(ctx,
uint32(command.PingCode), nil); err != nil {
- if !isReconnectable(err) {
- return nil, err
- }
- if _, err := c.do(ctx, &command.Ping{}); err != nil {
- return nil, err
- }
+ return nil, err
}
}
-
+ stamped, refreshed, repeatedPrimary := false, false, false
+ lifecycleRetryInterval := replayInterval
+ var requestParent *consumerSession
+ exact := captured != nil
for {
if err := ctx.Err(); err != nil {
return nil, err
}
- route, err := c.consumerRoute(ctx, key, routePayload, routeCode)
- var response []byte
+ if requestParent != nil && !c.matchesPollParent(*requestParent)
{
+ return nil, ierror.ErrUnauthenticated
+ }
+ state := pollExchangeState{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) {
+ return nil, ierror.ErrUnauthenticated
+ }
if captured == nil {
value := route.context
captured = &value
}
route.context = *captured
- response, err = c.pollOnRoute(ctx, code, key, payload,
route)
+ if !exact {
+ route.context.MetadataOp =
max(route.context.MetadataOp, route.parent.watermark)
+ }
+ if !stamped {
+ snapshot := c.pollSession.Load()
+ 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 {
Review Comment:
nit: the id is stamped from the shared counter before the per-endpoint gate
at :337 (master stamped inside the gate). Two goroutines on one partition can
send N+1 before N. The slice window still runs N, but with the counter shared
across all connections the 128-id margin shrinks with fanout, and the residual
is an explicit `RequestTooOld`. The LifecycleBusy restamp (:274) has the same
shape.
Stamp after the gate is acquired.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java:
##########
@@ -89,33 +95,105 @@ final class PollRouter {
}
CompletableFuture<ByteBuf> poll(ByteBuf payload,
Optional<PartitionContext> context) {
+ return partitionRequest(CommandCode.Messages.POLL.getValue(), payload,
context);
+ }
+
+ CompletableFuture<ByteBuf> partitionRequest(int command, ByteBuf payload,
Optional<PartitionContext> context) {
byte[] request = takeBytes(payload);
- return send(new Poll(
- CommandCode.Messages.GET_POLL_ROUTING,
- request,
- CommandCode.Messages.POLL_ON_PRIMARY.getValue(),
- request,
- request.length - POLL_PARAMETERS_BYTES,
- pollTimeout,
- context));
- }
-
- /**
- * Sends a store or delete offset request to the partition primary. The
route is
- * looked up by the consumer, stream, topic and partition in the first
- * {@code targetLength} bytes, as in Rust's {@code write_offset}.
- */
- CompletableFuture<ByteBuf> writeOffset(
- int command, ByteBuf payload, int targetLength,
Optional<PartitionContext> context) {
+ try {
+ byte[] query = partitionQuery(command, request);
+ return send(new Poll(query, command, request, pollTimeout,
context));
+ } catch (RuntimeException error) {
+ return CompletableFuture.failedFuture(error);
+ }
+ }
+
+ CompletableFuture<ByteBuf> sendMessages(ByteBuf payload, ByteBuf query) {
byte[] request = takeBytes(payload);
- return send(new Poll(
- CommandCode.ConsumerOffset.GET_ROUTING,
- Arrays.copyOf(request, targetLength),
- command,
- request,
- targetLength,
- pollTimeout,
- context));
+ byte[] target = takeBytes(query);
+ return send(new Poll(target, CommandCode.Messages.SEND.getValue(),
request, pollTimeout, Optional.empty()));
+ }
+
+ CompletableFuture<Long> partitionCount(StreamId stream, TopicId topic) {
+ ByteBuf query = BytesSerializer.toBytes(stream);
+ ByteBuf topicBytes = BytesSerializer.toBytes(topic);
+ try {
+ query.writeBytes(topicBytes);
+ query.writeIntLE(0);
+ } finally {
+ topicBytes.release();
+ }
+ byte[] target = takeBytes(query);
+ String topicKey = ByteBufUtil.hexDump(target, 0, target.length -
Integer.BYTES);
+ synchronized (this) {
+ Long count = partitionCounts.get(topicKey);
+ if (count != null) {
+ return CompletableFuture.completedFuture(count);
+ }
+ }
+ Poll lookup = new Poll(
+ target, CommandCode.Messages.GET_PARTITION_ROUTE.getValue(),
target, pollTimeout, Optional.empty());
+ return send(lookup).thenApply(response -> {
+ response.release();
+ return lookup.partitionsCount;
+ });
+ }
+
+ synchronized void clear() {
+ routes.clear();
+ partitionCounts.clear();
+ }
+
+ synchronized void observeTopology(ClusterMetadata metadata) {
+ replicas.clear();
+ rosterSize = metadata.nodes().size();
+ for (var node : metadata.nodes()) {
+ if (node.endpoints().tcp() != 0) {
+ replicas.put(
+ node.replicaId(),
+ new ConnectionInfo(node.ip(), node.endpoints().tcp()));
+ }
+ }
+ }
+
+ private static byte[] partitionQuery(int command, byte[] request) {
+ ByteBuf body = Unpooled.wrappedBuffer(request);
+ try {
+ int consumerKind = body.readUnsignedByte();
+ skipIdentifier(body);
+ int streamStart = body.readerIndex();
+ skipIdentifier(body);
+ skipIdentifier(body);
+ int partitionStart = body.readerIndex();
+ boolean explicit = body.readUnsignedByte() != 0;
+ long partition = body.readUnsignedIntLE();
+ if (!explicit) {
+ if (command == CommandCode.ConsumerOffset.STORE.getValue()
+ || command ==
CommandCode.ConsumerOffset.DELETE.getValue()
+ || (command == CommandCode.Messages.POLL.getValue() &&
consumerKind == 2)) {
+ throw new IggyInvalidArgumentException("A group poll or
offset mutation requires a partition id");
+ }
+ partition = 0;
+ body.setByte(partitionStart, 1);
+ body.setIntLE(partitionStart + 1, 0);
+ }
+ ByteBuf query = Unpooled.buffer(partitionStart - streamStart +
Integer.BYTES);
+ query.writeBytes(request, streamStart, partitionStart -
streamStart).writeIntLE((int) partition);
+ return takeBytes(query);
+ } catch (IndexOutOfBoundsException error) {
+ throw new IggyInvalidArgumentException("Truncated partition
target");
+ } finally {
+ body.release();
+ }
+ }
+
+ private static void skipIdentifier(ByteBuf body) {
+ int kind = body.readUnsignedByte();
+ int length = body.readUnsignedByte();
+ if ((kind == 1 && length != Integer.BYTES) || (kind != 1 && kind != 2)
|| length == 0) {
+ throw new IggyInvalidArgumentException("Invalid partition target
identifier");
+ }
+ body.skipBytes(length);
}
private static byte[] takeBytes(ByteBuf payload) {
Review Comment:
nit: every routed send now heap-copies the whole batch (single-node too)
plus a hexDump key and a scheduled timeout. In a cluster, the Slot tail chain
allows one send in flight per primary. Master pipelined on the main connection
via `retainedDuplicate`, though that clustered path never worked.
Keep the `ByteBuf` (`retainedDuplicate`) and pipeline SendMessages on the
data connection. Request identity doesn't need the serialization, since each
Poll owns its failover state.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs:
##########
@@ -201,105 +120,141 @@ private static IggyInvalidStatusCodeException
TruncatedSend()
return VsrError.Exception(VsrError.INVALID_COMMAND, "The SendMessages
body is truncated.");
}
- private static ContextQuery? TryBuildContextQuery(int code,
ReadOnlySpan<byte> body)
+ private static SendTarget PartitionTarget(int code, ReadOnlySpan<byte>
body)
{
- if (code is CommandCodes.STORE_CONSUMER_OFFSET_CODE or
CommandCodes.DELETE_CONSUMER_OFFSET_CODE)
+ if (code == CommandCodes.SEND_MESSAGES_CODE)
{
- var suffixSize = code == CommandCodes.STORE_CONSUMER_OFFSET_CODE
- ? StoreOffsetSuffixSize : DeleteOffsetSuffixSize;
- if (body.Length <= suffixSize)
- {
- return null;
- }
- var query = body[..^suffixSize].ToArray();
- return new
ContextQuery(CommandCodes.GET_CONSUMER_OFFSET_ROUTING_CODE, query,
ConsumerSessionSize,
- Convert.ToBase64String(query));
+ return ParseSendTarget(body);
}
- if (code != CommandCodes.POLL_MESSAGES_CODE || body.Length <=
PollOptionsSize)
+
+ if (body.IsEmpty)
{
- return null;
+ throw TruncatedSend();
+ }
+ var group = body[0] == ConsumerGroupWireKind;
+ var position = 1;
+ _ = ReadIdentifier(body, ref position);
+ var identifiersStart = position;
+ var streamId = ReadIdentifier(body, ref position);
+ var topicId = ReadIdentifier(body, ref position);
+ var identifiersLength = position - identifiersStart;
+ if (position + 1 + sizeof(uint) > body.Length)
+ {
+ throw TruncatedSend();
}
- return new ContextQuery(CommandCodes.GET_POLL_ROUTING_CODE,
body.ToArray(), ConsumerSessionSize,
- Convert.ToBase64String(body[..^PollOptionsSize]));
+ var present = body[position++] != 0;
+ if (!present && (code is CommandCodes.STORE_CONSUMER_OFFSET_CODE or
CommandCodes.DELETE_CONSUMER_OFFSET_CODE
+ || code == CommandCodes.POLL_MESSAGES_CODE && group))
+ {
+ throw VsrError.Exception(VsrError.INVALID_COMMAND, "This partition
request requires an explicit partition id.");
+ }
+ var partitionId = present ?
BinaryPrimitives.ReadUInt32LittleEndian(body[position..]) : 0;
+ var query = new byte[identifiersLength + sizeof(uint)];
+ body.Slice(identifiersStart, identifiersLength).CopyTo(query);
+
BinaryPrimitives.WriteUInt32LittleEndian(query.AsSpan(identifiersLength),
partitionId);
+ return new SendTarget(new PartitionRouteKey(streamId, topicId,
partitionId), query);
+ }
+
+ private static byte[] RouteQuery(PartitionRouteKey key)
+ {
+ var query = new byte[IdentifierPrefixSize * 2 + key.StreamId.Length +
key.TopicId.Length + sizeof(uint)];
+ query.AsSpan().WriteBytesFromStreamAndTopicIdentifiers(key.StreamId,
key.TopicId);
+ BinaryPrimitives.WriteUInt32LittleEndian(query.AsSpan(query.Length -
sizeof(uint)), key.PartitionId);
+ return query;
}
- /// <summary>
- /// Sends a poll or an offset write to the partition primary, over a
data connection that leaves the
- /// membership connection where it is. A caller context, or else the
first context a route reports, is
- /// retained across every retry, so a retry never moves the request
onto a newer incarnation or owner.
- /// </summary>
- /// <param name="code">The poll or offset write command code.</param>
- /// <param name="key">The cached route of the request.</param>
- /// <param name="routing">The body of the routing request.</param>
- /// <param name="payload">The body of the request.</param>
- /// <param name="context">The caller context, or null to take the context
the route reports.</param>
- /// <param name="token">The cancellation token.</param>
- private async Task<IMemoryOwner<byte>> SendRoutedAsync(int code,
PollRouteKey key, ReadOnlyMemory<byte> routing,
+ private async Task<IMemoryOwner<byte>> SendRoutedAsync(int code,
SendTarget target,
ReadOnlyMemory<byte> payload, PartitionContext? context,
CancellationToken token)
{
using var cancellation =
CancellationTokenSource.CreateLinkedTokenSource(token);
cancellation.CancelAfter(VsrRequestTimeoutMs);
var deadline = Environment.TickCount64 + VsrRequestTimeoutMs;
+ // A new route can change the destination, but not the request's
identity or captured position.
var captured = context;
+ SessionFrame? frame = null;
+ ulong? generation = null;
+ var refreshHistory = code == CommandCodes.SEND_MESSAGES_CODE &&
context is null;
+ var refusedPrimary = false;
+ var lifecycleInterval = LifecycleRetryIntervalMs;
try
{
while (true)
{
try
{
- if (Volatile.Read(ref _clusterNodeCount) == 0)
+ var route = await GetPartitionRouteAsync(target.Key,
target.Query, deadline, cancellation.Token);
+ captured ??= route.Route.Context;
+ if (generation is not null && generation !=
route.Generation)
{
- using var metadata = await
SendPollControlAsync(CommandCodes.GET_CLUSTER_METADATA_CODE,
- ReadOnlyMemory<byte>.Empty, deadline,
cancellation.Token);
- if (metadata.Memory.Length == 0)
- {
- throw new MalformedResponseException("Poll routing
requires a nonempty cluster roster.");
- }
-
-
RememberRoster(BinaryMapper.MapClusterMetadata(metadata.Memory.Span));
+ throw new VsrRequestOutcomeUnknownException(new
NotConnectedException());
Review Comment:
nit: a session generation change throws `VsrRequestOutcomeUnknownException`
even when nothing has been sent yet, whereas Node answers NotAccepted
(`client.socket.ts:768`). Only report an unknown outcome once a frame actually
went out.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java:
##########
@@ -182,102 +258,177 @@ private void attempt(Poll poll) {
}
return;
}
- synchronized (this) {
- routes.remove(poll.key);
- }
- if (isNotAccepted(error)) {
- if (!poll.retryRefusal()) {
- return;
- }
- coordinator
- .get()
- .eventLoop()
- .schedule(() -> attempt(poll), RETRY_INTERVAL_MILLIS,
TimeUnit.MILLISECONDS);
- } else {
- poll.result.completeExceptionally(unwrap(error));
- }
+ failedAttempt(poll, error);
});
}
+ private void failedAttempt(Poll poll, Throwable error) {
+ if (poll.retryingRefusal && unwrap(error) instanceof
IggyTimeoutException) {
+ poll.result.completeExceptionally(notAccepted());
+ return;
+ }
+ IggyServerException refusal =
AsyncTcpConnection.findServerError(error);
+ boolean historyRefresh = poll.canRefreshHistory(refusal);
+ boolean notPrimary = refusal != null && refusal.getRawErrorCode() ==
AsyncTcpConnection.NOT_PRIMARY;
+ invalidateRoute(poll, error, historyRefresh, notPrimary);
+ if (poll.isLifecycleRefusal(refusal)) {
+ retryLifecycle(poll, refusal);
+ return;
+ }
+ if (!poll.terminalRoutingRefusal && (isNotAccepted(error) ||
notPrimary || historyRefresh)) {
+ retry(poll, historyRefresh, notPrimary);
+ } else {
+ poll.result.completeExceptionally(unwrap(error));
+ }
+ }
+
+ private synchronized void invalidateRoute(Poll poll, Throwable error,
boolean historyRefresh, boolean notPrimary) {
+ if (historyRefresh) {
+ routes.keySet().removeIf(key -> key.startsWith(poll.topicKey()));
+ partitionCounts.remove(poll.topicKey());
+ } else if (isNotAccepted(error) || notPrimary || poll.dataCommand !=
CommandCode.Messages.SEND.getValue()) {
+ routes.remove(poll.key);
+ }
+ }
+
+ private void retryLifecycle(Poll poll, IggyServerException refusal) {
+ long delay = poll.lifecycleRetryMillis;
+ if (poll.deadline - System.nanoTime() <=
TimeUnit.MILLISECONDS.toNanos(delay)) {
+ poll.result.completeExceptionally(refusal);
+ return;
+ }
+ poll.lifecycleRefusal = refusal;
+ poll.state.renewRequestId();
+ poll.lifecycleRetryMillis = Math.min(delay * 2,
AsyncTcpConnection.LIFECYCLE_RETRY_MAX_INTERVAL_MS);
+ coordinator.get().eventLoop().schedule(() -> attempt(poll), delay,
TimeUnit.MILLISECONDS);
+ }
+
+ private void retry(Poll poll, boolean historyRefresh, boolean notPrimary) {
+ if (historyRefresh) {
+ poll.refreshedHistory = true;
+ poll.context.set(null);
+ poll.state = new AsyncTcpConnection.TransientFailoverState();
+ }
+ if (!poll.retryRefusal()) {
+ return;
+ }
+ long delay = (notPrimary && !poll.notPrimarySeen) || historyRefresh ?
0 : RETRY_INTERVAL_MILLIS;
+ poll.notPrimarySeen |= notPrimary;
+ coordinator.get().eventLoop().schedule(() -> attempt(poll), delay,
TimeUnit.MILLISECONDS);
+ }
+
private CompletableFuture<Route> route(Poll poll) {
AsyncTcpConnection parent = coordinator.get();
if (!parent.isAuthenticated()) {
return CompletableFuture.failedFuture(new
IggyNotConnectedException("Not authenticated, call login first"));
}
synchronized (this) {
- observeMetadata(parent.metadataWatermark());
Route cached = routes.get(poll.key);
if (cached != null && routeIsCurrent(cached)) {
return CompletableFuture.completedFuture(cached);
}
}
- return parent.send(poll.routingCommand,
Unpooled.wrappedBuffer(poll.routingBody))
- .thenApply(response -> decodeRoute(parent, poll, response))
- .exceptionallyCompose(error -> {
- if (!connectionFailed(error) || poll.result.isDone()) {
- return CompletableFuture.failedFuture(unwrap(error));
+ CompletableFuture<Void> roster = rosterSize == 0
+ ? refreshRoster(parent,
poll.deadline).exceptionallyCompose(error -> {
+ poll.terminalRoutingRefusal = true;
+ return CompletableFuture.failedFuture(unwrap(error));
+ })
+ : CompletableFuture.completedFuture(null);
+ return roster.thenCompose(ignored -> parent.sendRouted(
+ CommandCode.Messages.GET_PARTITION_ROUTE.getValue(),
+ Unpooled.wrappedBuffer(poll.routingBody),
+ 0,
+ poll.deadline,
+ new AsyncTcpConnection.TransientFailoverState(new
PartitionContext(
+ BigInteger.ZERO, BigInteger.ZERO,
BigInteger.valueOf(parent.metadataWatermark())))))
+ .thenCompose(response -> {
+ PartitionRoute decoded;
+ try {
+ decoded = PartitionRoute.decode(response);
+ } finally {
+ response.release();
+ }
+ CompletableFuture<Void> known;
+ synchronized (this) {
+ known = replicas.containsKey(decoded.primaryReplica())
|| connectData == null
+ ? CompletableFuture.completedFuture(null)
+ : refreshRoster(parent, poll.deadline);
}
- return coordinator
- .get()
- .send(CommandCode.System.PING,
Unpooled.EMPTY_BUFFER)
- .thenApply(response -> {
- response.release();
- throw notAccepted();
- });
+ return known.thenApply(ignored -> cacheRoute(parent, poll,
decoded));
});
}
- private Route decodeRoute(AsyncTcpConnection parent, Poll poll, ByteBuf
response) {
- try {
- if (response.readableBytes() < ATTACHMENT_BYTES +
PartitionContext.ENCODED_SIZE) {
- throw new IggyClientException("Truncated primary poll session
attachment");
- }
- Attachment attachment = new Attachment(
- response.readLongLE(), response.readLongLE(),
response.readLongLE(), response.readLongLE());
- long generation = parent.sessionGeneration();
- byte[] bindSecret = parent.bindSecret(attachment.clientLow,
attachment.clientHigh, attachment.session);
- if (generation != parent.sessionGeneration()) {
- throw new IggyClientException("Parent session changed while
decoding a poll route");
- }
- var context = BytesDeserializer.readPartitionContext(response);
- var node = BytesDeserializer.readClusterNode(response);
- if (response.isReadable() || node.endpoints().tcp() == 0) {
- throw new IggyClientException("Invalid TCP primary poll
routing response");
- }
- synchronized (this) {
- observeMetadata(parent.metadataWatermark());
- Route route = new Route(
- new ConnectionInfo(node.ip(), node.endpoints().tcp()),
- attachment.withWatermark(metadataWatermark),
- parent,
- generation,
- bindSecret,
- context);
- if (routes.size() >= MAX_ROUTES) {
- routes.clear();
- }
- if (!poll.result.isDone() && coordinator.get() == parent) {
- routes.put(poll.key, route);
- }
- return route;
- }
- } catch (IndexOutOfBoundsException | IggyMalformedResponseException
error) {
- throw new IggyClientException("Invalid TCP primary poll routing
response", error);
- } finally {
- response.release();
+ private CompletableFuture<Void> refreshRoster(AsyncTcpConnection parent,
long deadline) {
+ return parent.sendRouted(
+ CommandCode.System.GET_CLUSTER_METADATA.getValue(),
+ Unpooled.EMPTY_BUFFER,
+ 0,
+ deadline,
+ new AsyncTcpConnection.TransientFailoverState())
+ .thenAccept(response -> {
+ try {
+
observeTopology(BytesDeserializer.readClusterMetadata(response));
+ } finally {
+ response.release();
+ }
+ });
+ }
+
+ private synchronized Route cacheRoute(AsyncTcpConnection parent, Poll
poll, PartitionRoute decoded) {
+ ConnectionInfo endpoint = replicas.get(decoded.primaryReplica());
+ if (decoded.partitionsCount() == 0) {
+ throw new IggyMalformedResponseException("Partition route contains
no partitions");
}
+ if (rosterSize == 0 || (endpoint == null && connectData != null)) {
+ poll.terminalRoutingRefusal = true;
+ throw notAccepted();
+ }
+ long generation = parent.sessionGeneration();
+ long[] identity = parent.sessionIdentity();
+ Attachment attachment = new Attachment(
+ identity[0], identity[1], identity[2],
decoded.metadataOp().longValue());
+ byte[] secret = parent.bindSecret(identity[0], identity[1],
identity[2]);
+ Route route = new Route(
+ endpoint,
+ attachment,
+ parent,
+ generation,
+ secret,
+ decoded.context(),
+ decoded.partitionsCount(),
+ rosterSize <= 1 || connectData == null);
+ if (routes.size() >= MAX_ROUTES) {
+ clear();
+ }
+ if (coordinator.get() != parent || parent.sessionGeneration() !=
generation) {
+ throw notAccepted();
+ }
+ routes.put(poll.key, route);
+ partitionCounts.put(poll.topicKey(), decoded.partitionsCount());
+ return route;
}
private synchronized boolean routeIsCurrent(Route route) {
AsyncTcpConnection parent = coordinator.get();
- observeMetadata(parent.metadataWatermark());
- return route.parent == parent
- && route.generation == parent.sessionGeneration()
- && Long.compareUnsigned(route.attachment.watermark,
metadataWatermark) >= 0;
+ return route.parent == parent && route.generation ==
parent.sessionGeneration();
Review Comment:
nit: unlike Go (`tcp_poll_routing.go:431`), the cached route is not aged by
the metadata watermark (Node `client.socket.ts` is the same). After a partition
is recreated, one `HistoryUnavailable` surfaces on poll/offset calls before
re-routing.
Align with Go.
##########
foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs:
##########
@@ -27,6 +27,9 @@ public class ClusterNode
/// </summary>
public required string Name { get; set; }
+ /// <summary>Configured replica identity, independent of its position in
the cluster roster.</summary>
+ public byte ReplicaId { get; set; }
Review Comment:
nit: `ReplicaId` silently defaults to 0 when the field is absent, which
names the wrong replica (`node.rs:26-28` warns about exactly this). The Java
`ClusterNode.replicaId` primitive int has the same problem.
Make it `required` here, and boxed or failing on missing in Java.
##########
core/cli/src/commands/binary_cluster/get_cluster_metadata.rs:
##########
@@ -101,7 +103,8 @@ impl CliCommand for GetClusterMetadataCmd {
GetClusterMetadataOutput::List => {
cluster_metadata.nodes.iter().for_each(|node| {
event!(target: PRINT_TARGET, Level::INFO,
- "{}|{}|{}|{}|{}|{}|{}|{}",
+ "{}|{}|{}|{}|{}|{}|{}|{}|{}",
+ node.replica_id,
Review Comment:
nit: putting `replica_id` first shifts every `|`-separated field of the list
output and breaks scripts that parse it. Append it as the last column instead.
--
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]
