You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by li...@apache.org on 2026/08/10 20:48:47 UTC
(kafka) branch trunk updated: KAFKA-20232: Fix WakeupException in awaitMetadataUpdate() (#21592)
This is an automated email from the ASF dual-hosted git repository.
lianetm 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 93ee413aca8 KAFKA-20232: Fix WakeupException in awaitMetadataUpdate() (#21592)
93ee413aca8 is described below
commit 93ee413aca82031c028695679d14a671b2da14c9
Author: Brian Sang <sa...@gmail.com>
AuthorDate: Mon Aug 10 13:48:38 2026 -0700
KAFKA-20232: Fix WakeupException in awaitMetadataUpdate() (#21592)
Asynchronous offset commits may throw an unexpected WakeupException.
This was addressed for some code paths previously in
https://github.com/apache/kafka/pull/12626, but left out for the
`awaitMetadataUpdate()` code path. This patch fixes the problem by
passing through a flag to `awaitMetadataUpdate()` to indicate whether
wakeups should be disabled. This is used to disable wakeups in the
context of asynchronous offset commits. All other uses leave wakeups
enabled.
Reviewers: Nikita Shupletsov <ni...@shupletsov.ca>, Lianet Magrans
<lm...@confluent.io>
---
.../clients/consumer/internals/AbstractCoordinator.java | 2 +-
.../clients/consumer/internals/ConsumerNetworkClient.java | 13 ++++++++++++-
.../consumer/internals/AbstractCoordinatorTest.java | 15 +++++++++++++++
3 files changed, 28 insertions(+), 2 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index b38883e5efb..0d4ff6c0ea3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -308,7 +308,7 @@ public abstract class AbstractCoordinator implements Closeable {
if (future.isRetriable()) {
log.debug("Coordinator discovery failed, refreshing metadata", future.exception());
timer.sleep(retryBackoff.backoff(attempts++));
- client.awaitMetadataUpdate(timer);
+ client.awaitMetadataUpdate(timer, disableWakeup);
} else {
fatalException = future.exception();
log.info("FindCoordinator request hit fatal exception", fatalException);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
index 0ef184db0c7..5a9bba6ded3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
@@ -161,9 +161,20 @@ public class ConsumerNetworkClient implements Closeable {
* @return true if update succeeded, false otherwise.
*/
public boolean awaitMetadataUpdate(Timer timer) {
+ return awaitMetadataUpdate(timer, false);
+ }
+
+ /**
+ * Block waiting on the metadata refresh with a timeout.
+ *
+ * @param timer Timer bounding how long this method can block
+ * @param disableWakeup true if we should not check for wakeups, false otherwise
+ * @return true if update succeeded, false otherwise.
+ */
+ public boolean awaitMetadataUpdate(Timer timer, boolean disableWakeup) {
int version = this.metadata.requestUpdate(false);
do {
- poll(timer);
+ poll(timer, null, disableWakeup);
} while (this.metadata.updateVersion() == version && timer.notExpired());
return this.metadata.updateVersion() > version;
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index 2b26c9207cd..dd37130b531 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -323,6 +323,21 @@ public class AbstractCoordinatorTest {
);
}
+ @Test
+ public void testNoWakeupFromAsyncCoordinatorReadyOnRetriableError() {
+ setupCoordinator();
+ mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.COORDINATOR_NOT_AVAILABLE));
+ consumerClient.wakeup();
+ // The async variation should not throw WakeupException
+ coordinator.ensureCoordinatorReadyAsync();
+
+ consumerClient.wakeup();
+ mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.COORDINATOR_NOT_AVAILABLE));
+ assertThrows(WakeupException.class, () ->
+ coordinator.ensureCoordinatorReady(mockTime.timer(0))
+ );
+ }
+
@Test
public void testTimeoutAndRetryJoinGroupIfNeeded() throws Exception {
setupCoordinator();