heesung-sn commented on code in PR #23152:
URL: https://github.com/apache/pulsar/pull/23152#discussion_r1718569562
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java:
##########
@@ -325,6 +325,16 @@ public synchronized void start() throws
PulsarServerException {
ServiceUnitStateCompactionStrategy.class.getName()))
.create();
tableview.listen((key, value) -> handle(key, value));
+ tableview.forEach((serviceUnit, data) -> {
+ if (debug) {
+ log.info("Loaded the service unit state data. serviceUnit:
{}, data: {}", serviceUnit, data);
+ }
+ ServiceUnitState state = state(data);
+ if (state.equals(Owned) && isTargetBroker(data.dstBroker())) {
+ pulsar.getNamespaceService()
+
.onNamespaceBundleOwned(LoadManagerShared.getNamespaceBundle(pulsar,
serviceUnit));
+ }
+ });
Review Comment:
> I dont think we should use foreachAndListen because we dont want the
brokers to react on old messages. This can cause the unnecessary events in the
channel.
Let me give an example for this.
Currently, the topic has the following messages.
M1: bundle1, (Releasing from B1 to B2)
M2: bundle1, (Assigning to B2)
M3: bundle1, (Owned by B2)
When the topic does not run compaction yet, the table view can receive all
non-compacted existing messages, M1-M3, when starting the table view on B1.
Then, the broker B1 will unnecessarily close the Bundle bundle1 and send
duplicated message M2, if we pass 'handle()' to 'foreachAndListen'.
--
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]
