You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2026/06/09 15:15:07 UTC
(kafka) branch trunk updated: KAFKA-20673: AdminClient partition-leader APIs hang when a cached leader has left the cluster (#22493)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 182fc91e52d KAFKA-20673: AdminClient partition-leader APIs hang when a cached leader has left the cluster (#22493)
182fc91e52d is described below
commit 182fc91e52d774aca1333cbf5435f5d3f3fade0d
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Tue Jun 9 17:14:59 2026 +0200
KAFKA-20673: AdminClient partition-leader APIs hang when a cached leader has left the cluster (#22493)
Since the partition-leader cache fast path was introduced,
`AdminApiDriver` routes a cached leader straight to the fulfillment
stage with a `ConstantNodeIdProvider` for that broker id, skipping the
lookup stage. If the cached broker id is no longer present in the admin
client's cluster metadata — for example because the broker left the
cluster and was not replaced under the same id — the call can never be
assigned a node. `ConstantNodeIdProvider.provide()` returns null and the
call sits in `pendingCalls` issuing only broker-info metadata refreshes,
which never re-resolve the partition leader, so the request stays stuck
until `default.api.timeout.ms` expires and then fails with "Timed out
waiting for a node assignment". This affects all
`PartitionLeaderStrategy`-based APIs: `listOffsets`, `deleteRecords`,
`describeProducers` and `abortTransaction`.
The cache's existing staleness recovery only covers leader changes that
surface as per-partition errors such as `NOT_LEADER_OR_FOLLOWER`. There
is no path that handles a cached leader id simply disappearing from the
cluster without any such error reaching the admin client, so the stale
entry is believed until the deadline.
This change detects, on the admin client thread where metadata access is
safe, that a fulfillment call's target broker is absent from ready
metadata, and sends the affected keys back to the lookup stage so the
leader is re-resolved with a fresh topic metadata request. The liveness
check is performed on the admin client thread rather than where the
cache is read: the cache is consulted in the `AdminApiDriver`
constructor on the calling thread, where `AdminMetadataManager` (which
is confined to the admin client thread) cannot be accessed safely.
The behaviour is covered by a driver-level test that verifies a cached
fulfillment request is sent back to the lookup stage, and by an
end-to-end test in `KafkaAdminClientTest` that seeds the cache, removes
the cached leader from the cluster, and asserts that the subsequent
`listOffsets` re-resolves the leader and completes instead of timing
out.
Reviewers: Andrew Schofield <as...@confluent.io>
---
.../kafka/clients/admin/KafkaAdminClient.java | 33 ++++++
.../clients/admin/internals/AdminApiDriver.java | 27 +++++
.../kafka/clients/admin/KafkaAdminClientTest.java | 111 +++++++++++++++++++++
.../admin/internals/AdminApiDriverTest.java | 19 ++++
.../PartitionLeaderStrategyIntegrationTest.java | 37 +++++++
5 files changed, 227 insertions(+)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 0f59264b7ea..a02b4e7bcf2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -284,6 +284,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
@@ -957,6 +958,15 @@ public class KafkaAdminClient extends AdminClient {
runnable.pendingCalls.add(this);
}
+ /**
+ * Invoked by the polling loop when no node could be assigned to this call. Returns true if
+ * the call took corrective action and should be removed from the pending queue, false to
+ * remain pending and retry node assignment on a later iteration.
+ */
+ boolean handleNodeUnavailable(long now) {
+ return false;
+ }
+
private void handleTimeoutFailure(long now, Throwable cause) {
if (log.isDebugEnabled()) {
log.debug("{} timed out at {} after {} attempt(s)", this, now, tries,
@@ -1226,6 +1236,8 @@ public class KafkaAdminClient extends AdminClient {
call.curNode = node;
getOrCreateListValue(callsToSend, node).add(call);
return true;
+ } else if (call.handleNodeUnavailable(now)) {
+ return true;
} else {
log.trace("Unable to assign {} to a node.", call);
return false;
@@ -5151,6 +5163,27 @@ public class KafkaAdminClient extends AdminClient {
super.maybeRetry(currentTimeMs, throwable);
}
}
+
+ @Override
+ boolean handleNodeUnavailable(long currentTimeMs) {
+ OptionalInt brokerId = spec.scope.destinationBrokerId();
+ // The fulfillment target broker is no longer present in the cluster metadata. This
+ // happens when a stale entry in the partition leader cache points at a broker that
+ // has since left the cluster. Send the keys back to the lookup stage so the leader
+ // can be re-resolved, rather than waiting for the request deadline to expire without
+ // ever issuing another lookup. maybeRetryLookup is a no-op (returns false) for
+ // strategies that target a fixed broker id, in which case we leave the call pending.
+ if (brokerId.isPresent()
+ && metadataManager.isReady()
+ && metadataManager.nodeById(brokerId.getAsInt()) == null
+ && driver.maybeRetryLookup(currentTimeMs, spec)) {
+ log.debug("Broker {} for {} is no longer in the cluster metadata; retrying lookup.",
+ brokerId.getAsInt(), spec.name);
+ maybeSendRequests(driver, currentTimeMs);
+ return true;
+ }
+ return false;
+ }
};
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
index e2d025dfaba..9e8899cb492 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
@@ -188,6 +188,33 @@ public class AdminApiDriver<K, V> {
keys.forEach(this::unmap);
}
+ /**
+ * Send the keys of a fulfillment request back to the Lookup stage. This is invoked when a
+ * fulfillment request cannot be routed because its target broker is no longer present in the
+ * cluster metadata, for example when a stale entry in the partition leader cache pointed at a
+ * broker that has since left the cluster. Re-running the lookup gives us a chance to discover
+ * the current leader. Without this, such a request would remain unassignable until the request
+ * deadline expires.
+ *
+ * <p>This is a no-op for lookup strategies that target a fixed broker (i.e. the lookup scope
+ * carries a destination broker id, as with {@link StaticBrokerStrategy}), since {@link #unmap}
+ * would simply remap the keys straight back to the same fulfillment broker without making any
+ * progress. In that case this returns {@code false} and the request is left untouched, so the
+ * caller can leave it pending rather than retrying in a loop and exhausting the retry budget.
+ *
+ * @return true if the keys were moved back to the Lookup stage, false if no lookup is possible
+ */
+ public boolean maybeRetryLookup(long currentTimeMs, RequestSpec<K> spec) {
+ boolean canLookup = spec.keys.stream()
+ .anyMatch(key -> handler.lookupStrategy().lookupScope(key).destinationBrokerId().isEmpty());
+ if (!canLookup) {
+ return false;
+ }
+ clearInflightRequest(currentTimeMs, spec);
+ retryLookup(spec.keys);
+ return true;
+ }
+
/**
* Complete the future associated with the given key. After this is called, all keys will
* be taken out of both the Lookup and Fulfillment stages so that request are not retried.
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 84297246266..57610437012 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -173,6 +173,8 @@ import org.apache.kafka.common.quota.ClientQuotaAlteration;
import org.apache.kafka.common.quota.ClientQuotaEntity;
import org.apache.kafka.common.quota.ClientQuotaFilter;
import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.AddRaftVoterRequest;
import org.apache.kafka.common.requests.AddRaftVoterResponse;
import org.apache.kafka.common.requests.AlterClientQuotasResponse;
@@ -297,8 +299,10 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -8365,6 +8369,113 @@ public class KafkaAdminClientTest {
}
}
+ /**
+ * Reproduces the scenario where the partition leader cache holds an entry pointing at a broker
+ * that has since left the cluster (for example after a broker is recycled with a new id). The
+ * cached leader sends the request straight to the fulfillment stage, but the admin client can
+ * never route it because the broker is no longer in the metadata. Without re-running the lookup,
+ * the call would sit unassigned until the request deadline expires and fail with
+ * "Timed out waiting for a node assignment". The admin client should instead re-resolve the
+ * leader and complete the request.
+ */
+ @Test
+ public void testListOffsetsRetriesLookupWhenCachedLeaderLeavesCluster() throws Exception {
+ Node node0 = new Node(0, "localhost", 8120);
+ Node node1 = new Node(1, "localhost", 8121);
+ final TopicPartition tp0 = new TopicPartition("foo", 0);
+
+ // Initially foo-0 is led by node1.
+ final Cluster initialCluster = new Cluster("mockClusterId", asList(node0, node1),
+ singletonList(new PartitionInfo("foo", 0, node1, new Node[]{node0, node1}, new Node[]{node0, node1})),
+ emptySet(), emptySet(), node0);
+ // After node1 leaves the cluster, foo-0 is led by node0.
+ final Cluster shrunkCluster = new Cluster("mockClusterId", singletonList(node0),
+ singletonList(new PartitionInfo("foo", 0, node0, new Node[]{node0}, new Node[]{node0})),
+ emptySet(), emptySet(), node0);
+
+ MockTime time = new MockTime();
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, initialCluster,
+ newStrMap(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000",
+ AdminClientConfig.METADATA_MAX_AGE_CONFIG, "50"))) {
+ MockClient mockClient = env.kafkaClient();
+ mockClient.setNodeApiVersions(NodeApiVersions.create());
+
+ // First call: the lookup resolves foo-0 to node1 and caches it, then the offsets fetch
+ // succeeds on node1.
+ mockClient.prepareResponse(body -> body instanceof MetadataRequest,
+ prepareMetadataResponse(initialCluster, Errors.NONE));
+ mockClient.prepareResponseFrom(listOffsetsResponse(tp0, 100L), node1);
+ assertEquals(100L, env.adminClient().listOffsets(singletonMap(tp0, OffsetSpec.latest()))
+ .all().get().get(tp0).offset());
+
+ // Drop node1 from the admin client's metadata via the periodic broker-info refresh.
+ // Waiting for a second refresh guarantees the first one has been fully processed.
+ AtomicInteger refreshes = new AtomicInteger();
+ TestUtils.waitForCondition(() -> {
+ time.sleep(20);
+ if (respondToBrokerInfoRefresh(mockClient, shrunkCluster))
+ refreshes.incrementAndGet();
+ return refreshes.get() >= 2;
+ }, "Timed out waiting for the broker-info metadata refresh to drop node1");
+
+ // Second call: the cache still points foo-0 at node1, which is gone. The admin client
+ // must re-resolve the leader (now node0) rather than getting stuck until the deadline.
+ ListOffsetsResult result = env.adminClient().listOffsets(singletonMap(tp0, OffsetSpec.latest()));
+ TestUtils.waitForCondition(() -> {
+ time.sleep(20);
+ // Keep node1 out of the metadata, re-resolve foo-0 to node0, and satisfy the fetch.
+ respondToBrokerInfoRefresh(mockClient, shrunkCluster);
+ respondToTopicMetadata(mockClient, "foo", shrunkCluster);
+ respondToListOffsets(mockClient, tp0, 200L, node0);
+ return result.all().isDone();
+ }, "Timed out waiting for listOffsets to recover after the cached leader left the cluster");
+
+ assertEquals(200L, result.all().get().get(tp0).offset());
+ }
+ }
+
+ private static ListOffsetsResponse listOffsetsResponse(TopicPartition tp, long offset) {
+ return new ListOffsetsResponse(new ListOffsetsResponseData().setTopics(singletonList(
+ ListOffsetsResponse.singletonListOffsetsTopicResponse(tp, Errors.NONE, -1L, offset, 5))));
+ }
+
+ /**
+ * Respond out of order to the first in-flight request matching {@code matcher} with
+ * {@code response}. Returns true if a matching request was found and answered.
+ */
+ private static boolean respondToInFlightRequest(MockClient mockClient,
+ Predicate<ClientRequest> matcher,
+ AbstractResponse response) {
+ for (ClientRequest request : mockClient.requests()) {
+ if (matcher.test(request)) {
+ mockClient.respondToRequest(request, response);
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean respondToBrokerInfoRefresh(MockClient mockClient, Cluster cluster) {
+ return respondToInFlightRequest(mockClient, request -> {
+ AbstractRequest body = request.requestBuilder().build();
+ return body instanceof MetadataRequest && ((MetadataRequest) body).topics().isEmpty();
+ }, prepareMetadataResponse(cluster, Errors.NONE));
+ }
+
+ private static boolean respondToTopicMetadata(MockClient mockClient, String topic, Cluster cluster) {
+ return respondToInFlightRequest(mockClient, request -> {
+ AbstractRequest body = request.requestBuilder().build();
+ return body instanceof MetadataRequest && ((MetadataRequest) body).topics().contains(topic);
+ }, prepareMetadataResponse(cluster, Errors.NONE));
+ }
+
+ private static boolean respondToListOffsets(MockClient mockClient, TopicPartition tp, long offset, Node node) {
+ return respondToInFlightRequest(mockClient,
+ request -> request.requestBuilder().build() instanceof ListOffsetsRequest
+ && request.destination().equals(node.idString()),
+ listOffsetsResponse(tp, offset));
+ }
+
@Test
public void testListOffsetsRetriableErrors() throws Exception {
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
index 4ec9c87970f..ddceeedd189 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
@@ -189,6 +189,25 @@ class AdminApiDriverTest {
ctx.poll(emptyMap(), emptyMap());
}
+ @Test
+ public void testRetryLookupIsNoOpForFixedBrokerStrategy() {
+ // For a lookup strategy that targets a fixed broker, a fulfillment request whose broker has
+ // left the cluster cannot be re-resolved by another lookup. maybeRetryLookup must report it
+ // made no progress (return false) and leave the key mapped, so the caller does not spin
+ // retrying it and exhaust the retry budget.
+ TestContext ctx = TestContext.staticMapped(map("foo", 1));
+ ctx.handler.expectRequest(Set.of("foo"), completed("foo", 1L));
+
+ List<RequestSpec<String>> requestSpecs = ctx.driver.poll();
+ assertEquals(1, requestSpecs.size());
+ RequestSpec<String> spec = requestSpecs.get(0);
+ assertEquals(OptionalInt.of(1), spec.scope.destinationBrokerId());
+
+ assertFalse(ctx.driver.maybeRetryLookup(ctx.time.milliseconds(), spec));
+ // The key remains mapped to its fixed broker.
+ assertEquals(OptionalInt.of(1), ctx.driver.keyToBrokerId("foo"));
+ }
+
@Test
public void testFulfillmentFailure() {
TestContext ctx = TestContext.staticMapped(map(
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/PartitionLeaderStrategyIntegrationTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/PartitionLeaderStrategyIntegrationTest.java
index f5d471b8c55..7950d6025c6 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/PartitionLeaderStrategyIntegrationTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/PartitionLeaderStrategyIntegrationTest.java
@@ -357,6 +357,43 @@ public class PartitionLeaderStrategyIntegrationTest {
assertTrue(result.all().get(tp1).isDone());
}
+ @Test
+ public void testRetryLookupForStaleCachedLeader() {
+ // A leader cached from an earlier request can point at a broker that has since left the
+ // cluster. Such a request skips the lookup stage and goes straight to fulfillment, but the
+ // admin client can never route it because the broker is gone. The driver must be able to
+ // send the request back to the lookup stage so the leader can be re-resolved.
+ PartitionLeaderCache partitionLeaderCache = new PartitionLeaderCache();
+
+ TopicPartition tp0 = new TopicPartition("T", 0);
+
+ // Seed the cache so the request goes straight to the fulfillment stage on broker 1.
+ partitionLeaderCache.put(Map.of(tp0, 1));
+
+ PartitionLeaderStrategy.PartitionLeaderFuture<Void> result =
+ new PartitionLeaderStrategy.PartitionLeaderFuture<>(Collections.singleton(tp0), partitionLeaderCache);
+ AdminApiDriver<TopicPartition, Void> driver = buildDriver(result);
+
+ List<AdminApiDriver.RequestSpec<TopicPartition>> requestSpecs = driver.poll();
+ assertEquals(1, requestSpecs.size());
+
+ AdminApiDriver.RequestSpec<TopicPartition> spec = requestSpecs.get(0);
+ assertEquals(OptionalInt.of(1), spec.scope.destinationBrokerId());
+ assertEquals(Collections.singleton(tp0), spec.keys);
+
+ // Send the request back to the lookup stage. The leader can be re-resolved, so this reports
+ // that it moved the key back to lookup.
+ assertTrue(driver.maybeRetryLookup(time.milliseconds(), spec));
+ assertFalse(result.all().get(tp0).isDone());
+
+ requestSpecs = driver.poll();
+ assertEquals(1, requestSpecs.size());
+
+ AdminApiDriver.RequestSpec<TopicPartition> retrySpec = requestSpecs.get(0);
+ assertEquals(OptionalInt.empty(), retrySpec.scope.destinationBrokerId());
+ assertEquals(Collections.singleton(tp0), retrySpec.keys);
+ }
+
@Test
public void testFatalLookupError() {
TopicPartition tp0 = new TopicPartition("T", 0);