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 2024/10/10 13:53:40 UTC
(kafka) branch 3.8 updated: KAFKA-17731: Removed timed waiting signal for client telemetry close (#17431)
This is an automated email from the ASF dual-hosted git repository.
lianetm pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new 07c4f4c8364 KAFKA-17731: Removed timed waiting signal for client telemetry close (#17431)
07c4f4c8364 is described below
commit 07c4f4c83647cfa61b184e9f4e81632f561c45d6
Author: Apoorv Mittal <ap...@gmail.com>
AuthorDate: Thu Oct 10 12:19:05 2024 +0100
KAFKA-17731: Removed timed waiting signal for client telemetry close (#17431)
Reviewers: Andrew Schofield <as...@confluent.io>, Kirk True <kt...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>, Lianet Magrans <lm...@confluent.io>
---
.../consumer/internals/AsyncKafkaConsumer.java | 2 +-
.../consumer/internals/LegacyKafkaConsumer.java | 2 +-
.../kafka/clients/producer/KafkaProducer.java | 4 ++-
.../internals/ClientTelemetryReporter.java | 24 ++++---------
.../telemetry/internals/ClientTelemetrySender.java | 4 +--
.../internals/ClientTelemetryReporterTest.java | 42 ++++++++++++++++++++++
6 files changed, 54 insertions(+), 24 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 6622cb286d9..09473b2224e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -1227,7 +1227,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
wakeupTrigger.disableWakeups();
final Timer closeTimer = time.timer(timeout);
- clientTelemetryReporter.ifPresent(reporter -> reporter.initiateClose(timeout.toMillis()));
+ clientTelemetryReporter.ifPresent(ClientTelemetryReporter::initiateClose);
closeTimer.update();
// Prepare shutting down the network thread
swallow(log, Level.ERROR, "Failed to release assignment before closing consumer",
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java
index 641bc81d8a2..ba02c9e3868 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java
@@ -1125,7 +1125,7 @@ public class LegacyKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
AtomicReference<Throwable> firstException = new AtomicReference<>();
final Timer closeTimer = createTimerForRequest(timeout);
- clientTelemetryReporter.ifPresent(reporter -> reporter.initiateClose(timeout.toMillis()));
+ clientTelemetryReporter.ifPresent(ClientTelemetryReporter::initiateClose);
closeTimer.update();
// Close objects with a timeout. The timeout is required because the coordinator & the fetcher send requests to
// the server in the process of closing which may not respect the overall timeout defined for closing the
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 21476143199..2b2db730814 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -1383,6 +1383,9 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
} else {
// Try to close gracefully.
final Timer closeTimer = time.timer(timeout);
+ clientTelemetryReporter.ifPresent(ClientTelemetryReporter::initiateClose);
+ closeTimer.update();
+
if (this.sender != null) {
this.sender.initiateClose();
closeTimer.update();
@@ -1397,7 +1400,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
closeTimer.update();
}
}
- clientTelemetryReporter.ifPresent(reporter -> reporter.initiateClose(closeTimer.remainingMs()));
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
index fd2a3c695a1..f6790b0baf1 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
@@ -217,10 +217,10 @@ public class ClientTelemetryReporter implements MetricsReporter {
telemetryProvider.updateLabels(labels);
}
- public void initiateClose(long timeoutMs) {
+ public void initiateClose() {
log.debug("Initiate close of ClientTelemetryReporter");
try {
- clientTelemetrySender.initiateClose(timeoutMs);
+ clientTelemetrySender.initiateClose();
} catch (Exception exception) {
log.error("Failed to initiate close of client telemetry reporter", exception);
}
@@ -270,7 +270,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final Condition subscriptionLoaded = lock.writeLock().newCondition();
- private final Condition terminalPushInProgress = lock.writeLock().newCondition();
/*
Initial state should be subscription needed which should allow issuing first telemetry
request of get telemetry subscription.
@@ -601,8 +600,8 @@ public class ClientTelemetryReporter implements MetricsReporter {
}
@Override
- public void initiateClose(long timeoutMs) {
- log.debug("initiate close for client telemetry, check if terminal push required. Timeout {} ms.", timeoutMs);
+ public void initiateClose() {
+ log.debug("initiate close for client telemetry, check if terminal push required.");
lock.writeLock().lock();
try {
@@ -622,14 +621,7 @@ public class ClientTelemetryReporter implements MetricsReporter {
return;
}
- try {
- log.info("About to wait {} ms. for terminal telemetry push to be submitted", timeoutMs);
- if (!terminalPushInProgress.await(timeoutMs, TimeUnit.MILLISECONDS)) {
- log.info("Wait for terminal telemetry push to be submitted has elapsed; may not have actually sent request");
- }
- } catch (InterruptedException e) {
- log.warn("Error during client telemetry close", e);
- }
+ log.debug("Updated state to send terminal telemetry push request");
} finally {
lock.writeLock().unlock();
}
@@ -826,13 +818,9 @@ public class ClientTelemetryReporter implements MetricsReporter {
ClientTelemetryState oldState = state;
state = oldState.validateTransition(newState);
log.debug("Setting telemetry state from {} to {}", oldState, newState);
-
- if (newState == ClientTelemetryState.TERMINATING_PUSH_IN_PROGRESS) {
- terminalPushInProgress.signalAll();
- }
return true;
} catch (IllegalStateException e) {
- log.warn("Error updating client telemetry state, disabled telemetry", e);
+ log.warn("Error updating client telemetry state, disabled telemetry");
enabled = false;
return false;
} finally {
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
index e6bb9d40e73..99a50ecebd9 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
@@ -107,8 +107,6 @@ public interface ClientTelemetrySender extends AutoCloseable {
/**
* Initiates shutdown of this client. This method is called when the enclosing client instance
* is being closed. This method should not throw an exception if the client is already closed.
- *
- * @param timeoutMs The maximum time to wait for the client to close.
*/
- void initiateClose(long timeoutMs);
+ void initiateClose();
}
diff --git a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
index 0cb30c24ccb..64fc02d071a 100644
--- a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
@@ -728,6 +728,48 @@ public class ClientTelemetryReporterTest {
assertTrue(timeMs >= 500 && timeMs <= 1500);
}
+ @Test
+ public void testTelemetryReporterInitiateClose() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
+
+ clientTelemetryReporter.initiateClose();
+ assertEquals(ClientTelemetryState.TERMINATING_PUSH_NEEDED, ((ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter
+ .telemetrySender()).state());
+ }
+
+ @Test
+ public void testTelemetryReporterInitiateCloseNoSubscription() {
+ clientTelemetryReporter.initiateClose();
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, ((ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter
+ .telemetrySender()).state());
+ }
+
+ @Test
+ public void testTelemetryReporterInitiateCloseAlreadyInTerminatedStates() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.TERMINATING_PUSH_NEEDED));
+
+ clientTelemetryReporter.initiateClose();
+ assertEquals(ClientTelemetryState.TERMINATING_PUSH_NEEDED, ((ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter
+ .telemetrySender()).state());
+
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.TERMINATING_PUSH_IN_PROGRESS));
+ clientTelemetryReporter.initiateClose();
+ assertEquals(ClientTelemetryState.TERMINATING_PUSH_IN_PROGRESS, ((ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter
+ .telemetrySender()).state());
+
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.TERMINATED));
+ clientTelemetryReporter.initiateClose();
+ assertEquals(ClientTelemetryState.TERMINATED, ((ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter
+ .telemetrySender()).state());
+ }
+
@AfterEach
public void tearDown() {
clientTelemetryReporter.close();