[KAFKA-8877] Race condition on partition counter

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 2.2.1
• Component/s: producer
• Assignee: Unassigned
• Reporter: Oleg Kuznetsov
• Created: 05/Sep/19 08:12
• Updated: 05/Sep/19 12:36

Description

In the method:

org.apache.kafka.clients.producer.internals.DefaultPartitioner#nextValue

private int nextValue(String topic) {
    AtomicInteger counter = topicCounterMap.get(topic);
    if (null == counter) {
        counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
        AtomicInteger currentCounter = topicCounterMap.putIfAbsent(topic, counter);
        if (currentCounter != null) {
            counter = currentCounter;
        }
    }
    return counter.getAndIncrement();
}

the counter might be created multiple times instead of once.

I propose to replace it with something like topicCounterMap.compute(topic, _ -> ... (init the counter once per topic))  **