Skip to content

Commit 29f0459

Browse files
jfallowsclaude
jfallows
and
claude
authored
fix(binding-kafka): back off and reconnect proactively on NOT_LEADER_OR_FOLLOWER (#2555)
Port #2551 (support/1.x) to develop. KafkaCacheClientFetchFactory's cache_client fanout previously tore down every fetch member and forced the application to reconnect whenever the broker reported NOT_LEADER_OR_FOLLOWER. It now backs off with capped exponential delay and reconnects the downstream leg internally, keeping app-facing members open. KafkaCacheClientMetaFactory notifies both the fetch and produce fanouts as soon as a meta refresh resolves a new leader, letting a pending reconnect cut its backoff short. The onClientFanoutMemberOpening leader-mismatch guard no longer fires while the fanout's leader is unknown, so it doesn't tear down members that raced in during the backoff window. KafkaCacheClientProduceFactory already carried do this backoff/retry scaffolding; this port adds the missing reconnectAt guard on doClientFanInitialBeginIfNecessary and its own onLeaderReady hook so it benefits from the same proactive meta-driven reconnect. partition.not.leader.reconnect's client script now reflects the transparent reconnect (no visible reset/second connect). The original two-hop, application-driven reconnect contract is preserved under new scenario names (partition.not.leader.client.reconnect for the CacheFetchIT/FetchIT client-reconnect-delay-0 case, and partition.not.leader.cache.reconnect for the spec-level peer-to-peer counterpart), each with a corresponding IT method. Verified via full `mvnw verify` (checkstyle, license, unit + IT) on runtime/binding-kafka and specs/binding-kafka.spec: 635 tests, 0 failures. Claude-Session: https://claude.ai/code/session_01FbdHyarRhpWZ7cqGWrHAzJ Co-authored-by: Claude <noreply@anthropic.com>
1 parent 621aaa4 commit 29f0459

10 files changed

Lines changed: 606 additions & 41 deletions

File tree

‎runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientFetchFactory.java‎

Lines changed: 106 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,9 @@
2727
import static io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaOffsetType.LIVE;
2828
import static io.aklivity.zilla.runtime.engine.budget.BudgetCreditor.NO_BUDGET_ID;
2929
import static io.aklivity.zilla.runtime.engine.budget.BudgetDebitor.NO_DEBITOR_INDEX;
30+
import static io.aklivity.zilla.runtime.engine.concurrent.Signaler.NO_CANCEL_ID;
31+
import static java.lang.System.currentTimeMillis;
32+
import static java.util.concurrent.TimeUnit.SECONDS;
3033

3134
import java.util.ArrayList;
3235
import java.util.List;
@@ -111,6 +114,7 @@ public final class KafkaCacheClientFetchFactory implements BindingHandler
111114
private static final int FLAG_NONE = 0x00;
112115

113116
private static final int SIGNAL_FANOUT_REPLY_WINDOW = 1;
117+
private static final int SIGNAL_RECONNECT = 2;
114118

115119
private final BeginFW beginRO = new BeginFW();
116120
private final FlushFW flushRO = new FlushFW();
@@ -122,6 +126,7 @@ public final class KafkaCacheClientFetchFactory implements BindingHandler
122126
private final ExtensionFW extensionRO = new ExtensionFW();
123127
private final KafkaBeginExFW kafkaBeginExRO = new KafkaBeginExFW();
124128
private final KafkaFlushExFW kafkaFlushExRO = new KafkaFlushExFW();
129+
private final KafkaResetExFW kafkaResetExRO = new KafkaResetExFW();
125130

126131
private final BeginFW.Builder beginRW = new BeginFW.Builder();
127132
private final DataFW.Builder dataRW = new DataFW.Builder();
@@ -146,13 +151,15 @@ public final class KafkaCacheClientFetchFactory implements BindingHandler
146151
private final BindingHandler streamFactory;
147152
private final LongUnaryOperator supplyInitialId;
148153
private final LongUnaryOperator supplyReplyId;
154+
private final LongSupplier supplyTraceId;
149155
private final LongFunction<String> supplyNamespace;
150156
private final LongFunction<String> supplyLocalName;
151157
private final LongFunction<BudgetDebitor> supplyDebitor;
152158
private final LongFunction<KafkaBindingConfig> supplyBinding;
153159
private final Function<String, KafkaCache> supplyCache;
154160
private final LongFunction<KafkaCacheRoute> supplyCacheRoute;
155161
private final KafkaCacheCursorFactory cursorFactory;
162+
private final int reconnectDelay;
156163

157164
public KafkaCacheClientFetchFactory(
158165
KafkaConfiguration config,
@@ -170,12 +177,14 @@ public KafkaCacheClientFetchFactory(
170177
this.streamFactory = context.streamFactory();
171178
this.supplyInitialId = context::supplyInitialId;
172179
this.supplyReplyId = context::supplyReplyId;
180+
this.supplyTraceId = context::supplyTraceId;
173181
this.supplyNamespace = context::supplyNamespace;
174182
this.supplyLocalName = context::supplyLocalName;
175183
this.supplyBinding = supplyBinding;
176184
this.supplyDebitor = supplyDebitor;
177185
this.supplyCache = supplyCache;
178186
this.supplyCacheRoute = supplyCacheRoute;
187+
this.reconnectDelay = config.cacheServerReconnect();
179188
this.cursorFactory = new KafkaCacheCursorFactory(context.writeBuffer().capacity());
180189
}
181190

@@ -222,6 +231,7 @@ public MessageConsumer newStream(
222231
final long partitionOffset = progress.partitionOffset();
223232
final KafkaCacheRoute cacheRoute = supplyCacheRoute.apply(resolvedId);
224233
final long partitionKey = cacheRoute.topicPartitionKey(topicName, partitionId);
234+
final Int2IntHashMap leadersByPartitionId = cacheRoute.supplyLeadersByPartitionId(topicName);
225235

226236
KafkaCacheClientFetchFanout fanout = cacheRoute.clientFetchFanoutsByTopicPartition.get(partitionKey);
227237
if (fanout == null)
@@ -240,6 +250,7 @@ public MessageConsumer newStream(
240250
resolvedId,
241251
authorization,
242252
affinity,
253+
leadersByPartitionId,
243254
partition,
244255
defaultOffset);
245256

@@ -251,7 +262,6 @@ public MessageConsumer newStream(
251262
final KafkaFilterCondition condition = cursorFactory.asCondition(filters, evaluation);
252263
final long latestOffset = kafkaFetchBeginEx.partition().latestOffset();
253264
final KafkaOffsetType maximumOffset = KafkaOffsetType.valueOf((byte) latestOffset);
254-
final Int2IntHashMap leadersByPartitionId = cacheRoute.supplyLeadersByPartitionId(topicName);
255265
final int leaderId = leadersByPartitionId.get(partitionId);
256266

257267
newStream = new KafkaCacheClientFetchStream(
@@ -517,6 +527,7 @@ final class KafkaCacheClientFetchFanout
517527
private final long originId;
518528
private final long routedId;
519529
private final long authorization;
530+
private final Int2IntHashMap leadersByPartitionId;
520531
private final KafkaCachePartition partition;
521532
private final List<KafkaCacheClientFetchStream> members;
522533

@@ -538,18 +549,22 @@ final class KafkaCacheClientFetchFanout
538549
private long partitionOffset;
539550
private long stableOffset;
540551
private long latestOffset;
552+
private long reconnectAt = NO_CANCEL_ID;
553+
private int reconnectAttempt;
541554

542555
private KafkaCacheClientFetchFanout(
543556
long originId,
544557
long routedId,
545558
long authorization,
546559
long leaderId,
560+
Int2IntHashMap leadersByPartitionId,
547561
KafkaCachePartition partition,
548562
long defaultOffset)
549563
{
550564
this.originId = originId;
551565
this.routedId = routedId;
552566
this.authorization = authorization;
567+
this.leadersByPartitionId = leadersByPartitionId;
553568
this.partition = partition;
554569
this.partitionOffset = defaultOffset;
555570
this.stableOffset = DEFAULT_STABLE_OFFSET;
@@ -563,7 +578,7 @@ private void onClientFanoutMemberOpening(
563578
long traceId,
564579
KafkaCacheClientFetchStream member)
565580
{
566-
if (member.leaderId != leaderId && member.leaderId != LEADER_UNKNOWN)
581+
if (leaderId != LEADER_UNKNOWN && member.leaderId != leaderId && member.leaderId != LEADER_UNKNOWN)
567582
{
568583
doClientFanoutInitialAbortIfNecessary(traceId);
569584
doClientFanoutReplyResetIfNecessary(traceId);
@@ -603,11 +618,28 @@ private void onClientFanoutMemberClosed(
603618

604619
if (members.isEmpty())
605620
{
621+
if (reconnectAt != NO_CANCEL_ID)
622+
{
623+
signaler.cancel(reconnectAt);
624+
this.reconnectAt = NO_CANCEL_ID;
625+
}
626+
606627
doClientFanoutInitialAbortIfNecessary(traceId);
607628
doClientFanoutReplyResetIfNecessary(traceId);
608629
}
609630
}
610631

632+
void onLeaderReady(
633+
long traceId)
634+
{
635+
if (leaderId == LEADER_UNKNOWN && reconnectAt != NO_CANCEL_ID)
636+
{
637+
signaler.cancel(reconnectAt);
638+
this.reconnectAt = NO_CANCEL_ID;
639+
doClientFanoutInitialBeginIfNecessary(traceId);
640+
}
641+
}
642+
611643
private void doClientFanoutInitialBeginIfNecessary(
612644
long traceId)
613645
{
@@ -622,9 +654,38 @@ private void doClientFanoutInitialBeginIfNecessary(
622654
replyMax = 0;
623655
}
624656

625-
if (!KafkaState.initialOpening(state))
657+
if (!KafkaState.initialOpening(state) && reconnectAt == NO_CANCEL_ID)
626658
{
627-
doClientFanoutInitialBegin(traceId);
659+
if (leaderId == LEADER_UNKNOWN)
660+
{
661+
leaderId = leadersByPartitionId.get(partition.id());
662+
}
663+
664+
if (leaderId != LEADER_UNKNOWN)
665+
{
666+
doClientFanoutInitialBegin(traceId);
667+
}
668+
else
669+
{
670+
doClientFanoutDiscoverLeaderIfNecessary(traceId);
671+
}
672+
}
673+
}
674+
675+
private void doClientFanoutDiscoverLeaderIfNecessary(
676+
long traceId)
677+
{
678+
if (reconnectDelay != 0 && !members.isEmpty())
679+
{
680+
if (reconnectAt != NO_CANCEL_ID)
681+
{
682+
signaler.cancel(reconnectAt);
683+
}
684+
685+
this.reconnectAt = signaler.signalAt(
686+
currentTimeMillis() + Math.min(50 << reconnectAttempt++, SECONDS.toMillis(reconnectDelay)),
687+
SIGNAL_RECONNECT,
688+
this::onClientFanoutSignal);
628689
}
629690
}
630691

@@ -814,13 +875,48 @@ private void onClientFanoutInitialReset(
814875
final long traceId = reset.traceId();
815876
final OctetsFW extension = reset.extension();
816877

817-
members.forEach(s -> s.doClientInitialResetIfNecessary(traceId, extension));
818-
members.forEach(s -> s.doClientReplyAbortIfNecessary(traceId));
819-
members.clear();
820-
821878
state = KafkaState.closedInitial(state);
822879

823880
doClientFanoutReplyResetIfNecessary(traceId);
881+
882+
final KafkaResetExFW kafkaResetEx = extension.get(kafkaResetExRO::tryWrap);
883+
final int error = kafkaResetEx != null ? kafkaResetEx.error() : -1;
884+
885+
if (error == ERROR_NOT_LEADER_FOR_PARTITION)
886+
{
887+
leaderId = LEADER_UNKNOWN;
888+
}
889+
890+
if (reconnectDelay != 0 && !members.isEmpty() && KafkaError.of(error).isRetriable())
891+
{
892+
if (reconnectAt != NO_CANCEL_ID)
893+
{
894+
signaler.cancel(reconnectAt);
895+
}
896+
897+
this.reconnectAt = signaler.signalAt(
898+
currentTimeMillis() + Math.min(50 << reconnectAttempt++, SECONDS.toMillis(reconnectDelay)),
899+
SIGNAL_RECONNECT,
900+
this::onClientFanoutSignal);
901+
}
902+
else
903+
{
904+
members.forEach(s -> s.doClientInitialResetIfNecessary(traceId, extension));
905+
members.forEach(s -> s.doClientReplyAbortIfNecessary(traceId));
906+
members.clear();
907+
}
908+
}
909+
910+
private void onClientFanoutSignal(
911+
int signalId)
912+
{
913+
assert signalId == SIGNAL_RECONNECT;
914+
915+
this.reconnectAt = NO_CANCEL_ID;
916+
917+
final long traceId = supplyTraceId.getAsLong();
918+
919+
doClientFanoutInitialBeginIfNecessary(traceId);
824920
}
825921

826922
private void doClientFanoutInitialEndIfNecessary(
@@ -846,6 +942,8 @@ private void onClientFanoutInitialWindow(
846942
{
847943
if (!KafkaState.initialOpened(state))
848944
{
945+
this.reconnectAttempt = 0;
946+
849947
final long traceId = window.traceId();
850948

851949
state = KafkaState.openedInitial(state);

‎runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientMetaFactory.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@
3131
import io.aklivity.zilla.runtime.binding.kafka.internal.cache.KafkaCacheTopic;
3232
import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaBindingConfig;
3333
import io.aklivity.zilla.runtime.binding.kafka.internal.config.KafkaRouteConfig;
34+
import io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaCacheClientFetchFactory.KafkaCacheClientFetchFanout;
35+
import io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaCacheClientProduceFactory.KafkaCacheClientProduceFan;
3436
import io.aklivity.zilla.runtime.binding.kafka.internal.types.ArrayFW;
3537
import io.aklivity.zilla.runtime.binding.kafka.internal.types.Flyweight;
3638
import io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaPartitionFW;
@@ -649,6 +651,25 @@ private void onMetaFanoutReplyData(
649651
leadersByPartitionId.clear();
650652
partitions.forEach(p -> leadersByPartitionId.put(p.partitionId(), p.leaderId()));
651653

654+
partitions.forEach(p ->
655+
{
656+
final long partitionKey = cacheRoute.topicPartitionKey(topic.name(), p.partitionId());
657+
658+
final KafkaCacheClientFetchFanout fetchFanout =
659+
cacheRoute.clientFetchFanoutsByTopicPartition.get(partitionKey);
660+
if (fetchFanout != null)
661+
{
662+
fetchFanout.onLeaderReady(traceId);
663+
}
664+
665+
final KafkaCacheClientProduceFan produceFan =
666+
cacheRoute.clientProduceFansByTopicPartition.get(partitionKey);
667+
if (produceFan != null)
668+
{
669+
produceFan.onLeaderReady(traceId);
670+
}
671+
});
672+
652673
members.forEach(s -> s.doMetaReplyDataIfNecessary(traceId, kafkaDataEx));
653674
}
654675

‎runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -631,6 +631,17 @@ private void onClientFanMemberClosed(
631631
}
632632
}
633633

634+
void onLeaderReady(
635+
long traceId)
636+
{
637+
if (reconnectAt != NO_CANCEL_ID)
638+
{
639+
signaler.cancel(reconnectAt);
640+
this.reconnectAt = NO_CANCEL_ID;
641+
doClientFanInitialBeginIfNecessary(traceId);
642+
}
643+
}
644+
634645
private void doClientFanInitialBeginIfNecessary(
635646
long traceId)
636647
{
@@ -639,7 +650,7 @@ private void doClientFanInitialBeginIfNecessary(
639650
state = 0;
640651
}
641652

642-
if (!KafkaState.initialOpening(state))
653+
if (!KafkaState.initialOpening(state) && reconnectAt == NO_CANCEL_ID)
643654
{
644655
doClientFanInitialBegin(traceId);
645656
}

‎runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheFetchIT.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,17 @@ public void shouldReconnectPartitionNotLeader() throws Exception
113113
k3po.finish();
114114
}
115115

116+
@Test
117+
@Configuration("cache.yaml")
118+
@Specification({
119+
"${app}/partition.not.leader.client.reconnect/client",
120+
"${app}/partition.not.leader.client.reconnect/server"})
121+
@ScriptProperty("serverAddress \"zilla://streams/app1\"")
122+
public void shouldClientReconnectPartitionNotLeader() throws Exception
123+
{
124+
k3po.finish();
125+
}
126+
116127
@Test
117128
@Configuration("cache.yaml")
118129
@Specification({

0 commit comments

Comments
 (0)