[
https://issues.apache.org/jira/browse/KAFKA-21243?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128399#comment-18128399
]
Viktor Klimin commented on KAFKA-21243:
---------------------------------------
Updated steps to reproduce - previous version had less concurrency and was
harder t reproduce with short local RTT inside localhost
{code:java}
public static void main(String[] a) throws Exception {
String topic = "probe";
ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
int probes = 5000;
long pauseMs = 8_000, stallMs = 400;
long estimateMs = 40; // initial guess of the metadata round
trip
Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
p.put("key.serializer", ByteArraySerializer.class.getName());
p.put("value.serializer", ByteArraySerializer.class.getName());
p.put("metadata.max.idle.ms", "5000"); // minimum Kafka allows
p.put("metadata.max.age.ms", "1000"); // refresh often so eviction
happens inside the pause
p.put("retry.backoff.ms", "750");
// + security settings if needed
KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(p);
byte[] payload = "x".getBytes();
for (int cycle = 0; cycle < 11; cycle++) {
CountDownLatch startingGun = new CountDownLatch(1);
List<Future<Long>> fs = new ArrayList<>();
Future<Long> trigger = pool.submit(() -> {
startingGun.await(); // Wait for the signal
long t = System.nanoTime();
producer.send(new ProducerRecord<>(topic, payload)).get();
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - t);
});
for (int i = 0; i < probes; i++) {
fs.add(pool.submit(() -> {
startingGun.await();
long t = System.nanoTime();
producer.send(new ProducerRecord<>(topic, payload)).get();
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - t);
}));
}
startingGun.countDown();
long triggerMs = trigger.get();
List<Long> stalls = new ArrayList<>();
for (Future<Long> f : fs) {
long ms = f.get();
if (ms >= stallMs) stalls.add(ms);
}
boolean evicted = triggerMs >= 5;
System.out.printf("cycle=%d evicted=%b trigger=%dms stalledProbes=%d
%s%n",
cycle, evicted, triggerMs, stalls.size(), stalls);
if (cycle > 0 && triggerMs >= 5 && triggerMs < stallMs)
estimateMs = (estimateMs * 3 + triggerMs) / 4;
Thread.sleep(pauseMs);
}
pool.shutdownNow();
producer.close();
} {code}
> Producer: concurrent first sends can wait an extra retry.backoff.ms for a
> redundant metadata refresh
> ----------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21243
> URL: https://issues.apache.org/jira/browse/KAFKA-21243
> Project: Kafka
> Issue Type: Bug
> Components: producer
> Affects Versions: 3.9.2, 4.4.1
> Reporter: Viktor Klimin
> Assignee: Edmond Abraham
> Priority: Major
>
> h3. Issue
> We run a producer with metadata.max.idle.ms at its default (5 min) and
> retry.backoff.ms=750. When a topic has been idle longer than
> metadata.max.idle.ms and several threads call send() at about the same time,
> some of them occasionally block for roughly 900 ms inside send(). That is
> about one retry.backoff.ms, even though the brokers are healthy and the first
> metadata response comes back in a couple of milliseconds. In our case two
> threads that called send() within ~2 ms of each other were both stuck for
> ~897 ms and were released together.
> h3. Description
> # In KafkaProducer.{*}waitOnMetadata{*}() the thread first takes a snapshot
> with metadata.fetch(), then calls metadata.requestUpdateForTopic(topic).
> These are two separate lock acquisitions. If the sender thread applies the
> first metadata response in between: ProducerMetadata.update() removes the
> topic from newTopics, bumps updateVersion (V -> V+1), and sets lastRefreshMs
> = now.
> # The waiting thread still has the old snapshot (no partitions for the
> topic), so it does not take the fast path.
> # It calls {*}requestUpdateForTopic{*}(topic). The topic is no longer in
> newTopics, so this goes to {*}requestUpdate{*}(false). That only sets
> needFullUpdate and returns V+1.
> # {*}awaitUpdate{*}(V+1) now needs a second metadata response, but a new
> refresh is not allowed until *lastRefreshMs + retry.backoff.ms* (with
> jitter), so the thread sits there for ~{*}0.6-0.9 s at 750{*} ms.
> The data the thread needs is already in the metadata at step 2, so the wait
> is redundant. When the second response arrives, all the threads that hit this
> window wake up together.
> If requestUpdateForTopic() happens before the response is applied, the topic
> is still in newTopics, requestUpdateForNewTopics() is used (no backoff), and
> the thread is released by the first response. So only the interleaving
> decides.
> h3. Reproduction
> The race is also reproducible with a real KafkaProducer.
> Test setup: one producer, metadata.max.idle.ms=5000,
> metadata.max.age.ms=1000, retry.backoff.ms=750, brokers ~35 ms away. Every
> cycle: wait until the topic is evicted from the producer metadata, send from
> one "trigger" thread, then send from 100 other threads at randomized instants
> spread over a ~20 ms window around the time the first metadata response is
> expected. Plain simultaneous start does not hit it, because all threads then
> arrive before the first response is applied; the arrivals have to be spread
> around the moment the response is applied.
> Since the window is the time between {*}fetch{*}() and
> {*}requestUpdateForTopic{*}(), the example below shows the same.
> {code:java}
> public class MetadataRaceProbe {
> public static void main(String[] a) throws Exception {
> String topic = "probe";
> int probes = 100;
> long spreadMs = 20, pauseMs = 7_000, stallMs = 400;
> long estimateMs = 40; // initial guess of the metadata
> round trip
>
> Properties p = new Properties();
> p.put("bootstrap.servers", "broker:9092");
> p.put("key.serializer", ByteArraySerializer.class.getName());
> p.put("value.serializer", ByteArraySerializer.class.getName());
> p.put("metadata.max.idle.ms", "5000"); // minimum Kafka allows
> p.put("metadata.max.age.ms", "1000"); // refresh often so
> eviction happens inside the pause
> p.put("retry.backoff.ms", "750");
> // + security settings if needed
>
> KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(p);
> ExecutorService pool = Executors.newFixedThreadPool(probes + 1);
> byte[] payload = "x".getBytes();
>
> for (int cycle = 0; cycle < 85; cycle++) {
> long t0 = System.nanoTime() + 100_000_000L; // common
> start, 100 ms ahead
> long est = estimateMs * 1_000_000L, spread = spreadMs *
> 1_000_000L;
>
> Future<Long> trigger = pool.submit(timedSend(producer, topic,
> payload, t0));
> List<Future<Long>> fs = new ArrayList<>();
> long slot = spread / probes;
> for (int i = 0; i < probes; i++) {
> long offset = -spread / 2 + i * slot +
> ThreadLocalRandom.current().nextLong(slot);
> fs.add(pool.submit(timedSend(producer, topic, payload, t0 +
> Math.max(0, est + offset))));
> }
>
> long triggerMs = trigger.get();
> List<Long> stalls = new ArrayList<>();
> for (Future<Long> f : fs) { long ms = f.get(); if (ms >= stallMs)
> stalls.add(ms); }
>
> boolean evicted = triggerMs >= 5; // trigger paid a
> metadata round trip
> System.out.printf("cycle=%d evicted=%b trigger=%dms
> stalledProbes=%d %s%n",
> cycle, evicted, triggerMs, stalls.size(), stalls);
>
> if (cycle > 0 && triggerMs >= 5 && triggerMs < stallMs)
> estimateMs = (estimateMs * 3 + triggerMs) / 4; // follow
> the real RTT
> Thread.sleep(pauseMs); // > idle + age,
> so the topic gets evicted
> }
> pool.shutdownNow();
> producer.close();
> }
> static Callable<Long> timedSend(KafkaProducer<byte[], byte[]> producer,
> String topic, byte[] payload, long targetNanos) {
> return () -> {
> waitUntil(targetNanos);
> long t = System.nanoTime();
> Future<RecordMetadata> ack = producer.send(new
> ProducerRecord<>(topic, payload));
> long sendMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
> t);
> ack.get();
> return sendMs;
> };
> }
> static void waitUntil(long target) {
> long left;
> while ((left = target - System.nanoTime()) > 0) {
> if (left > 200_000) LockSupport.parkNanos(left - 150_000);
> else Thread.onSpinWait();
> }
> }
> }{code}
> h3. Expected behavior
> A thread whose snapshot is older than an already-applied metadata update
> should not schedule a new refresh. It should just re-read the metadata, find
> the topic, and carry on.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
