From 27b2f343a480911ac48514bd81c96ab949258b14 Mon Sep 17 00:00:00 2001 From: "xi.wang" Date: Fri, 10 Jul 2026 23:35:14 +0000 Subject: [PATCH 1/2] Reduce lock contention in ProducerMetadata#add MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit add() is called on every send() via KafkaProducer.waitOnMetadata, so the synchronized method contends with the Sender thread's update()/ retainTopic() calls on every produced record even though the vast majority of calls are just refreshing an already-tracked topic's expiry. Switch the topics map to a ConcurrentHashMap and refresh known topics via a lock-free replace(), which only writes if the topic is still present. For a topic seen for the first time (or one concurrently evicted by retainTopic()), fall back to a synchronized block so the map insert and newTopics bookkeeping happen atomically with retainTopic() - otherwise retainTopic() could expire-and-remove the topic in the gap between the two, leaving it recorded in newTopics without the corresponding map entry, or silently re-inserting an evicted topic without the newTopics bookkeeping needed to trigger an immediate refresh. retainTopic() switches to a conditional remove(topic, expireMs) so a concurrent refresh of an existing topic in add() can't be undone by an eviction decision based on the expiry value read just before the refresh landed. The batch add(Collection, long) overload is unaffected: it stays fully synchronized, remains mutually exclusive with retainTopic(), and only races benignly with the lock-free path (a last-write-wins refresh of the same timestamp semantics). It's used only by KafkaProducer#sendOffsetsToTransaction, not the per-record send() path, so it doesn't need the same treatment. Adds ProducerMetadataTest#testRetainTopic for direct coverage of retainTopic()'s branches, and a JMH benchmark (ProducerMetadataAddBenchmark) comparing add() against a baseline mirroring the prior fully-synchronized implementation. With 8 threads concurrently refreshing a shared pool of 200 already-tracked topics: Benchmark Mode Cnt Score Error Units ProducerMetadataAddBenchmark.addExistingTopicFullySynchronized thrpt 5 6217.287 ± 1279.054 ops/ms ProducerMetadataAddBenchmark.addExistingTopicLockFreeHotPath thrpt 5 27232.344 ± 5404.836 ops/ms ~4.4x higher throughput under this contention pattern. --- .../producer/internals/ProducerMetadata.java | 37 ++++-- .../internals/ProducerMetadataTest.java | 28 ++++ .../ProducerMetadataAddBenchmark.java | 122 ++++++++++++++++++ 3 files changed, 176 insertions(+), 11 deletions(-) create mode 100644 jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java index a35f02f6a130c..6945092c14331 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java @@ -29,19 +29,19 @@ import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Objects; import java.util.OptionalInt; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; public class ProducerMetadata extends Metadata { // If a topic hasn't been accessed for this many milliseconds, it is removed from the cache. private final long metadataIdleMs; /* Topics with expiry time */ - private final Map topics = new HashMap<>(); + private final Map topics = new ConcurrentHashMap<>(); private final Set newTopics = new HashSet<>(); private final Logger log; private final Time time; @@ -70,11 +70,21 @@ public synchronized MetadataRequest.Builder newMetadataRequestBuilderForNewTopic return new MetadataRequest.Builder(new ArrayList<>(newTopics), true); } - public synchronized void add(String topic, long nowMs) { + public void add(String topic, long nowMs) { Objects.requireNonNull(topic, "topic cannot be null"); - if (topics.put(topic, nowMs + metadataIdleMs) == null) { - newTopics.add(topic); - requestUpdateForNewTopics(); + long expiryTime = nowMs + metadataIdleMs; + // replace() only writes if the topic is still present, so there's no window in which a topic + // concurrently evicted by retainTopic() could get silently re-inserted without newTopics bookkeeping. + if (topics.replace(topic, expiryTime) != null) { + return; + } + synchronized (this) { + // New (or concurrently-evicted) topic: topics.put() and newTopics.add() must happen atomically + // with retainTopic(), hence the shared lock. + if (topics.put(topic, expiryTime) == null) { + newTopics.add(topic); + requestUpdateForNewTopics(); + } } } @@ -124,15 +134,20 @@ public synchronized boolean retainTopic(String topic, boolean isInternal, long n Long expireMs = topics.get(topic); if (expireMs == null) { return false; - } else if (newTopics.contains(topic)) { + } + if (newTopics.contains(topic)) { return true; - } else if (expireMs <= nowMs) { + } + if (expireMs > nowMs) { + return true; + } + // Only remove if the expiry we read is still current: the lock-free refresh path in add() can + // race with this check, and a plain remove(topic) would drop a concurrently-refreshed entry. + if (topics.remove(topic, expireMs)) { log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", topic, expireMs, nowMs); - topics.remove(topic); return false; - } else { - return true; } + return true; } /** diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java index 3353c64ff0f2f..01b202e29c82b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java @@ -304,6 +304,34 @@ public void testRequestUpdateForTopic() { assertFalse(metadata.updateRequested()); } + @Test + public void testRetainTopic() { + final String unknownTopic = "unknown-topic"; + final String newTopic = "new-topic"; + final String activeTopic = "active-topic"; + + // A topic that was never added isn't retained. + assertFalse(metadata.retainTopic(unknownTopic, false, 0)); + + // A "new" topic (added but not yet resolved by an update) is retained even past its + // nominal expiry, since it hasn't had a chance to be fetched yet. + metadata.add(newTopic, 0); + assertTrue(metadata.newTopics().contains(newTopic)); + assertTrue(metadata.retainTopic(newTopic, false, METADATA_IDLE_MS * 10)); + + // Once resolved (no longer "new"), a topic that hasn't reached its idle expiry is retained. + metadata.add(activeTopic, 0); + metadata.updateWithCurrentRequestVersion(responseWithTopics(Set.of(newTopic, activeTopic)), true, 0); + assertFalse(metadata.newTopics().contains(activeTopic)); + assertTrue(metadata.retainTopic(activeTopic, false, METADATA_IDLE_MS - 1)); + + // Past its idle expiry, the topic is evicted: retainTopic() returns false and actually + // removes the topic, so a subsequent call also returns false. + assertFalse(metadata.retainTopic(activeTopic, false, METADATA_IDLE_MS)); + assertFalse(metadata.containsTopic(activeTopic)); + assertFalse(metadata.retainTopic(activeTopic, false, METADATA_IDLE_MS)); + } + private MetadataResponse responseWithCurrentTopics() { return responseWithTopics(metadata.topics()); } diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java new file mode 100644 index 0000000000000..556b63c4e2c09 --- /dev/null +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.jmh.producer; + +import org.apache.kafka.clients.producer.internals.ProducerMetadata; +import org.apache.kafka.common.internals.ClusterResourceListeners; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.internals.LogContext; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +/** + * Compares {@link ProducerMetadata#add} against a baseline that mirrors the prior, fully + * {@code synchronized} implementation (map update + newTopics bookkeeping as a single + * {@code synchronized} block). Every {@code send()} call refreshes the expiry of the topic + * it's producing to via {@code add()}, so with many application threads sharing one producer, + * this refresh is on the hot path and contends across threads for a topic that's virtually + * always already tracked. This benchmark simulates that: many threads concurrently refreshing + * a shared pool of already-registered topics. + */ +@State(Scope.Benchmark) +@Fork(value = 1) +@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Threads(8) +public class ProducerMetadataAddBenchmark { + + private static final int TOPIC_COUNT = 200; + private static final long METADATA_IDLE_MS = TimeUnit.MINUTES.toMillis(5); + + private String[] topics; + private ProducerMetadata lockFreeHotPathMetadata; + private FullySynchronizedProducerMetadata fullySynchronizedMetadata; + + @Setup(Level.Trial) + public void setup() { + topics = new String[TOPIC_COUNT]; + for (int i = 0; i < TOPIC_COUNT; i++) { + topics[i] = "topic-" + i; + } + + lockFreeHotPathMetadata = new ProducerMetadata(100L, 1000L, TimeUnit.MINUTES.toMillis(5), METADATA_IDLE_MS, + new LogContext(), new ClusterResourceListeners(), Time.SYSTEM); + fullySynchronizedMetadata = new FullySynchronizedProducerMetadata(METADATA_IDLE_MS); + + long nowMs = Time.SYSTEM.milliseconds(); + for (String topic : topics) { + lockFreeHotPathMetadata.add(topic, nowMs); + fullySynchronizedMetadata.add(topic, nowMs); + } + } + + @Benchmark + public void addExistingTopicFullySynchronized() { + String topic = topics[ThreadLocalRandom.current().nextInt(TOPIC_COUNT)]; + fullySynchronizedMetadata.add(topic, Time.SYSTEM.milliseconds()); + } + + @Benchmark + public void addExistingTopicLockFreeHotPath() { + String topic = topics[ThreadLocalRandom.current().nextInt(TOPIC_COUNT)]; + lockFreeHotPathMetadata.add(topic, Time.SYSTEM.milliseconds()); + } + + /** + * Baseline mirroring the prior {@code ProducerMetadata#add}: a plain {@code HashMap} + * guarded end-to-end by a single {@code synchronized} method, so every call - including a + * refresh of an already-tracked topic - serializes on the instance lock. + */ + private static class FullySynchronizedProducerMetadata { + private final long metadataIdleMs; + private final Map topics = new HashMap<>(); + private final Set newTopics = new HashSet<>(); + + FullySynchronizedProducerMetadata(long metadataIdleMs) { + this.metadataIdleMs = metadataIdleMs; + } + + synchronized void add(String topic, long nowMs) { + Objects.requireNonNull(topic, "topic cannot be null"); + if (topics.put(topic, nowMs + metadataIdleMs) == null) { + newTopics.add(topic); + } + } + } +} From 0f47c11028c1721ce8f13fa341de7a3ad5f0f35f Mon Sep 17 00:00:00 2001 From: "xi.wang" Date: Mon, 13 Jul 2026 19:00:19 +0000 Subject: [PATCH 2/2] Simplify ProducerMetadataAddBenchmark to benchmark only add() Drop the hand-mirrored fully-synchronized baseline in favor of running this benchmark before/after a change to add() and comparing throughput directly, since a duplicated baseline implementation can drift out of sync with the real one it's meant to represent. --- .../ProducerMetadataAddBenchmark.java | 62 +++++-------------- 1 file changed, 14 insertions(+), 48 deletions(-) diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java index 556b63c4e2c09..1e5e1055574a3 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/producer/ProducerMetadataAddBenchmark.java @@ -35,22 +35,19 @@ import org.openjdk.jmh.annotations.Threads; import org.openjdk.jmh.annotations.Warmup; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Objects; -import java.util.Set; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; /** - * Compares {@link ProducerMetadata#add} against a baseline that mirrors the prior, fully - * {@code synchronized} implementation (map update + newTopics bookkeeping as a single - * {@code synchronized} block). Every {@code send()} call refreshes the expiry of the topic - * it's producing to via {@code add()}, so with many application threads sharing one producer, - * this refresh is on the hot path and contends across threads for a topic that's virtually - * always already tracked. This benchmark simulates that: many threads concurrently refreshing - * a shared pool of already-registered topics. + * Benchmarks {@link ProducerMetadata#add} refreshing an already-tracked topic's expiry. Every + * {@code send()} call does this refresh, so with many application threads sharing one producer, + * it's on the hot path and contends across threads for a topic that's virtually always already + * tracked. This benchmark simulates that: many threads concurrently refreshing a shared pool of + * already-registered topics. + * + * To measure the effect of a change to {@code add()}, run this benchmark before and after the + * change and compare the throughput, rather than hand-mirroring the old implementation here + * where it could drift out of sync with reality. */ @State(Scope.Benchmark) @Fork(value = 1) @@ -65,8 +62,7 @@ public class ProducerMetadataAddBenchmark { private static final long METADATA_IDLE_MS = TimeUnit.MINUTES.toMillis(5); private String[] topics; - private ProducerMetadata lockFreeHotPathMetadata; - private FullySynchronizedProducerMetadata fullySynchronizedMetadata; + private ProducerMetadata metadata; @Setup(Level.Trial) public void setup() { @@ -75,48 +71,18 @@ public void setup() { topics[i] = "topic-" + i; } - lockFreeHotPathMetadata = new ProducerMetadata(100L, 1000L, TimeUnit.MINUTES.toMillis(5), METADATA_IDLE_MS, + metadata = new ProducerMetadata(100L, 1000L, TimeUnit.MINUTES.toMillis(5), METADATA_IDLE_MS, new LogContext(), new ClusterResourceListeners(), Time.SYSTEM); - fullySynchronizedMetadata = new FullySynchronizedProducerMetadata(METADATA_IDLE_MS); long nowMs = Time.SYSTEM.milliseconds(); for (String topic : topics) { - lockFreeHotPathMetadata.add(topic, nowMs); - fullySynchronizedMetadata.add(topic, nowMs); + metadata.add(topic, nowMs); } } @Benchmark - public void addExistingTopicFullySynchronized() { - String topic = topics[ThreadLocalRandom.current().nextInt(TOPIC_COUNT)]; - fullySynchronizedMetadata.add(topic, Time.SYSTEM.milliseconds()); - } - - @Benchmark - public void addExistingTopicLockFreeHotPath() { + public void addExistingTopic() { String topic = topics[ThreadLocalRandom.current().nextInt(TOPIC_COUNT)]; - lockFreeHotPathMetadata.add(topic, Time.SYSTEM.milliseconds()); - } - - /** - * Baseline mirroring the prior {@code ProducerMetadata#add}: a plain {@code HashMap} - * guarded end-to-end by a single {@code synchronized} method, so every call - including a - * refresh of an already-tracked topic - serializes on the instance lock. - */ - private static class FullySynchronizedProducerMetadata { - private final long metadataIdleMs; - private final Map topics = new HashMap<>(); - private final Set newTopics = new HashSet<>(); - - FullySynchronizedProducerMetadata(long metadataIdleMs) { - this.metadataIdleMs = metadataIdleMs; - } - - synchronized void add(String topic, long nowMs) { - Objects.requireNonNull(topic, "topic cannot be null"); - if (topics.put(topic, nowMs + metadataIdleMs) == null) { - newTopics.add(topic); - } - } + metadata.add(topic, Time.SYSTEM.milliseconds()); } }