gnarula commented on code in PR #22810:
URL: https://github.com/apache/kafka/pull/22810#discussion_r3578163116
##########
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:
I suppose this check is because the overloaded `add(Collection<String>
topics, long nowMs)` might have mutated `topics`?
--
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]
