Tycho Schottdorf created KAFKA-21369:
----------------------------------------
Summary: 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: clients, admin
Affects Versions: 4.3.1, 4.0.0
Reporter: Tycho Schottdorf
*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)
