eeriee commented on code in PR #22810:
URL: https://github.com/apache/kafka/pull/22810#discussion_r3580696962
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java:
##########
@@ -70,11 +70,21 @@ public synchronized MetadataRequest.Builder
newMetadataRequestBuilderForNewTopic
return new MetadataRequest.Builder(new ArrayList<>(newTopics), true);
}
- public synchronized void add(String topic, long nowMs) {
+ public void add(String topic, long nowMs) {
Objects.requireNonNull(topic, "topic cannot be null");
- if (topics.put(topic, nowMs + metadataIdleMs) == null) {
- newTopics.add(topic);
- requestUpdateForNewTopics();
+ long expiryTime = nowMs + metadataIdleMs;
+ // replace() only writes if the topic is still present, so there's no
window in which a topic
+ // concurrently evicted by retainTopic() could get silently
re-inserted without newTopics bookkeeping.
+ if (topics.replace(topic, expiryTime) != null) {
+ return;
+ }
+ synchronized (this) {
+ // New (or concurrently-evicted) topic: topics.put() and
newTopics.add() must happen atomically
+ // with retainTopic(), hence the shared lock.
+ if (topics.put(topic, expiryTime) == null) {
Review Comment:
this is the same old behavior moving to a sync block, yeah the overloaded
add or the same add method can be called to update topic concurrently.
--
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]
