Demogorgon314 commented on code in PR #21668:
URL: https://github.com/apache/pulsar/pull/21668#discussion_r1414736153
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -651,6 +651,19 @@ public void addComplete(Position pos, ByteBuf entryData,
Object ctx) {
@Override
public synchronized void addFailed(ManagedLedgerException exception,
Object ctx) {
+ /* If the topic is being transferred(in the Releasing bundle state),
+ we don't want to forcefully close topic here.
+ Instead, we will rely on the service unit state channel's
bundle(topic) transfer protocol.
+ At the end of the transfer protocol, at Owned state, the source
broker should close the topic properly.
+ */
+ if (transferring) {
Review Comment:
Does this change relevant to the flaky test? If not, a separate PR might be
better.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java:
##########
@@ -157,6 +157,7 @@ public abstract class AbstractTopic implements Topic,
TopicPolicyListener<TopicP
protected final LongAdder msgOutFromRemovedSubscriptions = new LongAdder();
protected final LongAdder bytesOutFromRemovedSubscriptions = new
LongAdder();
protected volatile Pair<String, List<EntryFilter>> entryFilters;
+ protected volatile boolean transferring = false;
Review Comment:
Can we reuse `isFenced`?
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java:
##########
@@ -2661,6 +2661,27 @@ The delayed message index time step(in seconds) in per
bucket snapshot segment,
)
private boolean loadBalancerSheddingBundlesWithPoliciesEnabled = false;
+ @FieldContext(
+ category = CATEGORY_LOAD_BALANCER,
+ doc = "Time to wait before fixing any stuck in-flight service unit
states. "
+ + "The leader monitor fixes any in-flight service
unit(bundle) states "
+ + "by reassigning the ownerships if stuck too long, longer
than this period."
+ + "(only used in load balancer extension logics)"
+ )
+ private long loadBalancerInFlightServiceUnitStateWaitingTimeInMillis = 30
* 1000;
Review Comment:
Do we need a PIP to add configuration? /cc @codelipenghui
--
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]
