Fix publish buffer of one ServerCnx caculated multi-times when check broker publish buffer
Fixes
Master Issue: #<xyz>
### Motivation
This PR fix publish buffer of one ServerCnx caculated multi-times when check broker's publish buffer.
### Modifications
check publish buffer of broker by adding messagePublishBufferSize of a ServerCnx for only one time.
@@ -48,6 +48,7 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -2223,7 +2224,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
@VisibleForTesting
void checkMessagePublishBuffer() {
AtomicLong currentMessagePublishBufferBytes = new AtomicLong();
- foreachProducer(producer -> currentMessagePublishBufferBytes.addAndGet(producer.getCnx().getMessagePublishBufferSize()));
+ foreachCnx(cnx -> currentMessagePublishBufferBytes.addAndGet(cnx.getMessagePublishBufferSize()));
if (currentMessagePublishBufferBytes.get() >= maxMessagePublishBufferBytes
&& !reachMessagePublishBufferThreshold) {
reachMessagePublishBufferThreshold = true;
@@ -2236,11 +2237,13 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
}
}
- private void foreachProducer(Consumer<Producer> consumer) {
+ private void foreachCnx(Consumer<ServerCnx> consumer) {
+ Set<ServerCnx> cnxSet = new HashSet<>();
topics.forEach((n, t) -> {
Optional<Topic> topic = extractTopic(t);
- topic.ifPresent(value -> value.getProducers().values().forEach(consumer));
+ topic.ifPresent(value -> value.getProducers().values().forEach(producer -> cnxSet.add(producer.getCnx())));
});
+ cnxSet.forEach(consumer);
}
public boolean isReachMessagePublishBufferThreshold() {
@@ -2250,7 +2253,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
@VisibleForTesting
long getCurrentMessagePublishBufferSize() {
AtomicLong currentMessagePublishBufferBytes = new AtomicLong();
- foreachProducer(producer -> currentMessagePublishBufferBytes.addAndGet(producer.getCnx().getMessagePublishBufferSize()));
+ foreachCnx(cnx -> currentMessagePublishBufferBytes.addAndGet(cnx.getMessagePublishBufferSize()));
return currentMessagePublishBufferBytes.get();
}
