Viktor Klimin created KAFKA-21243:
-------------------------------------
Summary: 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
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)
