[jira] [Assigned] (KAFKA-21243) Producer: concurrent first sends can wait an extra retry.backoff.ms for a redundant metadata refresh

["Edmond Abraham (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-21243?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Edmond Abraham reassigned KAFKA-21243:
--------------------------------------

    Assignee: Edmond Abraham

> 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)