Skip to content

KAFKA-20804: Cut lock contention in metadata add()#22869

Closed
1230fahid wants to merge 1 commit into
apache:trunkfrom
1230fahid:fa-kafka-20804-lock-contention
Closed

KAFKA-20804: Cut lock contention in metadata add()#22869
1230fahid wants to merge 1 commit into
apache:trunkfrom
1230fahid:fa-kafka-20804-lock-contention

Conversation

@1230fahid

Copy link
Copy Markdown

Summary

ProducerMetadata.add() is called on every KafkaProducer.send() via
waitOnMetadata. Because it was fully synchronized, all producer threads
contended on the same instance lock even for topics already in the metadata
cache.

Changes

  • topics map switched to ConcurrentHashMap. add(String) and
    add(Collection) now take a lock-free fast path via
    ConcurrentHashMap.replace() for already-known topics, covering the hot
    path on every send(). Only truly new (or concurrently evicted) topics
    fall back to a synchronized block, keeping the map insert and newTopics
    bookkeeping atomic with retainTopic().

  • retainTopic() switches from a plain topics.remove(topic) to a
    conditional topics.remove(topic, expireMs). This prevents a race where
    retainTopic() reads a stale (expired) expiry, add() concurrently
    refreshes it via replace(), and retainTopic() then removes the freshly
    refreshed entry. The conditional remove loses the CAS if the value has
    changed, leaving the refreshed entry intact.

Testing

  • testAddKnownTopicFastPathDoesNotAddToNewTopics: verifies the fast path
    does not re-add a known topic to newTopics or trigger an update.
  • testBatchAddAllKnownTopicsReturnsEmptyOptional /
    testBatchAddWithNewTopicReturnsPresentOptional: verifies add(Collection)
    fast/slow path return values.
  • testConcurrentAddOfSameNewTopicIsIdempotent: 20 threads racing to add the
    same new topic and asserts newTopics contains it exactly once.
  • testRetainTopicConditionalRemovePreservesRefreshedTopic: 200 iterations of
    concurrent add() and retainTopic() racing at the expiry boundary and
    asserts the topic always survives when add() has refreshed its expiry.

This commit switches add() to use ConcurrentHashMap.replace() for
known topics, avoiding the instance lock on every send(). New
or evicted topics fall back to a synchronized block so map insert
and newTopics bookkeeping remain atomic with retainTopic().
retainTopic() also switches to conditional remove(topic, expireMs)
to prevent evicting a topic concurrently refreshed by the add()
fast path.
@github-actions github-actions Bot added triage PRs from the community producer clients labels Jul 19, 2026
@1230fahid
1230fahid marked this pull request as draft July 19, 2026 00:40
@1230fahid
1230fahid marked this pull request as ready for review July 19, 2026 00:41
@1230fahid

Copy link
Copy Markdown
Author

This is actually duplicate work for: #22810, so will close this PR.

@1230fahid 1230fahid closed this Jul 19, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

clients producer triage PRs from the community

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants