1230fahid opened a new pull request, #22869: URL: https://github.com/apache/kafka/pull/22869
## 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 is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
