[
https://issues.apache.org/jira/browse/KAFKA-21369?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Apoorv Mittal reassigned KAFKA-21369:
-------------------------------------
Assignee: Apoorv Mittal
> KafkaAdminClient calls MetricsReporter.close() twice during a single shutdown
> -----------------------------------------------------------------------------
>
> Key: KAFKA-21369
> URL: https://issues.apache.org/jira/browse/KAFKA-21369
> Project: Kafka
> Issue Type: Bug
> Components: admin, clients
> Affects Versions: 4.0.0, 4.3.1
> Reporter: Tycho Schottdorf
> Assignee: Apoorv Mittal
> Priority: Major
>
> *Affected versions verified at runtime:* 4.0.0 and 4.3.1. Control: 3.9.0. The
> same shutdown paths are present in trunk.
> h2. Description
> A single call to {{Admin.close(Duration)}} invokes {{close()}} twice on the
> same configured {{MetricsReporter}} instance. The calls originate from the
> caller thread and the Admin client's network thread. A producer configured
> with the same counting reporter closes it once.
> In Kafka 4.3.1,
> [KafkaAdminClient.close(Duration)|https://github.com/apache/kafka/blob/4.3.1/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L679]
> calls {{metrics.close()}}, and [AdminClientRunnable
> cleanup|https://github.com/apache/kafka/blob/4.3.1/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L1490]
> calls {{closeQuietly(metrics, "Metrics")}}.
> [Metrics.close()|https://github.com/apache/kafka/blob/4.3.1/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java#L674]
> invokes every reporter on each call without a closed-state guard.
> This affects reporters that release shared resources. With Strimzi
> client-metrics-reporter 0.4.0, an Admin client releases the shared HTTP
> listener twice and can stop the endpoint while a producer remains open.
> h2. Reproduction
> Save the following as {{ReporterCloseCount.java}}. It uses Kafka classes only
> and does not require a broker or Strimzi's reporter. The Strimzi container
> supplies the exact tested Kafka client jar and Java runtime.
> {code:java}
> import java.time.Duration;
> import java.util.HashMap;
> import java.util.List;
> import java.util.Map;
> import java.util.concurrent.atomic.AtomicInteger;
> import org.apache.kafka.clients.admin.Admin;
> import org.apache.kafka.clients.producer.KafkaProducer;
> import org.apache.kafka.common.metrics.KafkaMetric;
> import org.apache.kafka.common.metrics.MetricsReporter;
> import org.apache.kafka.common.serialization.StringSerializer;
> import org.apache.kafka.common.utils.AppInfoParser;
> public class ReporterCloseCount {
> public static class Counting implements MetricsReporter {
> private final AtomicInteger closes = new AtomicInteger();
> private String clientId;
> public void configure(Map<String, ?> config) {
> clientId = config.get("client.id").toString();
> }
> public void init(List<KafkaMetric> metrics) { }
> public void metricChange(KafkaMetric metric) { }
> public void metricRemoval(KafkaMetric metric) { }
> public void close() {
> System.out.printf("%s reporter=%x close=%d thread=%s%n",
> clientId, System.identityHashCode(this),
> closes.incrementAndGet(),
> Thread.currentThread().getName());
> }
> }
> static Map<String, Object> config(String id) {
> return new HashMap<>(Map.of(
> "bootstrap.servers", "127.0.0.1:1",
> "client.id", id,
> "metric.reporters", Counting.class.getName()));
> }
> public static void main(String[] args) {
>
> Thread.currentThread().setContextClassLoader(ReporterCloseCount.class.getClassLoader());
> System.out.println("Kafka " + AppInfoParser.getVersion());
> Map<String, Object> p = config("producer");
> p.put("key.serializer", StringSerializer.class.getName());
> p.put("value.serializer", StringSerializer.class.getName());
> KafkaProducer<String, String> producer = new KafkaProducer<>(p);
> producer.close(Duration.ofSeconds(5));
> Admin admin = Admin.create(config("admin"));
> admin.close(Duration.ofSeconds(5));
> }
> }
> {code}
> {code:bash}
> docker run --rm --network none --read-only --tmpfs /tmp:rw,nosuid,size=256m \
> -v "$PWD:/repro:ro" --entrypoint java \
>
> quay.io/strimzi/kafka:1.2.0-kafka-4.3.1@sha256:fef34b5438e8556cc08c01f3e254e47346f061b53a4e38d4289853777e0ea7f1
> \
> -cp "/opt/kafka/libs/*" /repro/ReporterCloseCount.java
> {code}
> Relevant output with Kafka 4.3.1 and Java 21.0.12:
> {noformat}
> Kafka 4.3.1
> producer reporter=28cb9120 close=1 thread=main
> admin reporter=5e742e4 close=1 thread=main
> admin reporter=5e742e4 close=2 thread=kafka-admin-client-thread | admin
> {noformat}
> The two Admin lines identify the same reporter instance. A separate version
> comparison using a counting reporter produced:
> || Kafka client version || Producer reporter close calls || Admin reporter
> close calls ||
> | 3.9.0 | 1 | 1 |
> | 4.0.0 | 1 | 2 |
> | 4.3.1 | 1 | 2 |
> h2. Expected behavior
> Closing an Admin client once should close each owned reporter once. Repeated
> closure of its {{Metrics}} repository should not invoke reporter teardown
> again.
> {{Metrics}} implements {{java.io.Closeable}}, whose [contract requires
> subsequent closes to have no
> effect|https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/io/Closeable.html].
> {{MetricsReporter}} extends {{AutoCloseable}}, whose [contract does not
> require
> idempotence|https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/lang/AutoCloseable.html].
> h2. Regression and related work
> The direct {{metrics.close()}} call was added in [commit
> 4ed0a958|https://github.com/apache/kafka/commit/4ed0a958e54ee6c3b80e4eb5fbc986e5b6be49e5]
> for [KAFKA-17248|https://issues.apache.org/jira/browse/KAFKA-17248],
> released in Kafka 4.0.0. The [original review
> discussion|https://github.com/apache/kafka/pull/17021#discussion_r1759041707]
> records that addition.
> [KAFKA-19504|https://issues.apache.org/jira/browse/KAFKA-19504] concerns
> duplicate reporter construction and leaked unused instances. This report
> concerns repeated closure of the same instance and reproduces on 4.3.1 after
> that fix.
> [Strimzi metrics-reporter
> #144|https://github.com/strimzi/metrics-reporter/issues/144] records the
> concrete shared-listener failure.
> h2. Possible resolution
> Ensure reporters are torn down only once during Admin shutdown, either by
> correcting shutdown ownership in {{KafkaAdminClient}} or by making
> {{Metrics.close()}} idempotent. A regression test should count calls to a
> custom reporter during a single Admin close and cover shutdown from both
> threads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
