You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2024/07/27 18:43:43 UTC
(pulsar) branch branch-3.1 updated: [fix][broker][branch-3.1] Do not try to clean owned bundles from inactive source brokers (ExtensibleLoadManagerImpl only) (#23064) (#23082)
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 42aec11403f [fix][broker][branch-3.1] Do not try to clean owned bundles from inactive source brokers (ExtensibleLoadManagerImpl only) (#23064) (#23082)
42aec11403f is described below
commit 42aec11403fd0541253a3a2d4fe8b62665139493
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Sat Jul 27 11:43:37 2024 -0700
[fix][broker][branch-3.1] Do not try to clean owned bundles from inactive source brokers (ExtensibleLoadManagerImpl only) (#23064) (#23082)
---
.../extensions/ExtensibleLoadManagerImpl.java | 27 ++-
.../channel/ServiceUnitStateChannelImpl.java | 195 ++++++++-------------
.../pulsar/broker/service/BrokerService.java | 9 +
.../channel/ServiceUnitStateChannelTest.java | 73 ++++++--
4 files changed, 161 insertions(+), 143 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
index 855e8ccfcaa..80fa59d4f7a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java
@@ -122,6 +122,9 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
private static final String ELECTION_ROOT = "/loadbalance/extension/leader";
+ private static final Set<String> INTERNAL_TOPICS =
+ Set.of(BROKER_LOAD_DATA_STORE_TOPIC, TOP_BUNDLES_LOAD_DATA_STORE_TOPIC, TOPIC);
+
private PulsarService pulsar;
private ServiceConfiguration conf;
@@ -756,7 +759,8 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
}
public static boolean isInternalTopic(String topic) {
- return topic.startsWith(TOPIC)
+ return INTERNAL_TOPICS.contains(topic)
+ || topic.startsWith(TOPIC)
|| topic.startsWith(BROKER_LOAD_DATA_STORE_TOPIC)
|| topic.startsWith(TOP_BUNDLES_LOAD_DATA_STORE_TOPIC);
}
@@ -914,5 +918,26 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
serviceUnitStateChannel.cleanOwnerships();
leaderElectionService.close();
brokerRegistry.unregister();
+ // Close the internal topics (if owned any) after giving up the possible leader role,
+ // so that the subsequent lookups could hit the next leader.
+ closeInternalTopics();
+ }
+
+ private void closeInternalTopics() {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (String name : INTERNAL_TOPICS) {
+ futures.add(pulsar.getBrokerService().getTopicIfExists(name)
+ .thenAccept(topicOptional -> topicOptional.ifPresent(topic -> topic.close(true)))
+ .exceptionally(__ -> {
+ log.warn("Failed to close internal topic:{}", name);
+ return null;
+ }));
+ }
+ try {
+ FutureUtil.waitForAll(futures)
+ .get(pulsar.getConfiguration().getNamespaceBundleUnloadingTimeoutMs(), TimeUnit.MILLISECONDS);
+ } catch (Throwable e) {
+ log.warn("Failed to wait for closing internal topics", e);
+ }
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index 1471d4a75c1..c18938c0fdb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -611,20 +611,13 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
private CompletableFuture<Void> publishOverrideEventAsync(String serviceUnit,
- ServiceUnitStateData orphanData,
ServiceUnitStateData override) {
if (!validateChannelState(Started, true)) {
throw new IllegalStateException("Invalid channel state:" + channelState.name());
}
EventType eventType = EventType.Override;
eventCounters.get(eventType).getTotal().incrementAndGet();
- return pubAsync(serviceUnit, override).whenComplete((__, e) -> {
- if (e != null) {
- eventCounters.get(eventType).getFailure().incrementAndGet();
- log.error("Failed to override serviceUnit:{} from orphanData:{} to overrideData:{}",
- serviceUnit, orphanData, override, e);
- }
- }).thenApply(__ -> null);
+ return pubAsync(serviceUnit, override).thenApply(__ -> null);
}
public CompletableFuture<Void> publishUnloadEventAsync(Unload unload) {
@@ -1241,45 +1234,51 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
broker, delayInSecs, cleanupJobs.size());
}
-
- private ServiceUnitStateData getOverrideInactiveBrokerStateData(ServiceUnitStateData orphanData,
- Optional<String> selectedBroker,
- String inactiveBroker) {
-
-
- if (selectedBroker.isEmpty()) {
- return new ServiceUnitStateData(Free, null, inactiveBroker,
- true, getNextVersionId(orphanData));
- }
-
- if (orphanData.state() == Splitting) {
- return new ServiceUnitStateData(Splitting, orphanData.dstBroker(), selectedBroker.get(),
- Map.copyOf(orphanData.splitServiceUnitToDestBroker()),
- true, getNextVersionId(orphanData));
- } else {
- return new ServiceUnitStateData(Owned, selectedBroker.get(), inactiveBroker,
- true, getNextVersionId(orphanData));
- }
- }
-
private void overrideOwnership(String serviceUnit, ServiceUnitStateData orphanData, String inactiveBroker) {
- Optional<String> selectedBroker = selectBroker(serviceUnit, inactiveBroker);
- if (selectedBroker.isEmpty()) {
- log.warn("Empty selected broker for ownership serviceUnit:{} orphanData:{}."
+ final var version = getNextVersionId(orphanData);
+ try {
+ selectBroker(serviceUnit, inactiveBroker)
+ .thenApply(selectedOpt ->
+ selectedOpt.map(selectedBroker -> {
+ if (orphanData.state() == Splitting) {
+ // if Splitting, set orphan.dstBroker() as dst to indicate where it was from.
+ // (The src broker runs handleSplitEvent.)
+ return new ServiceUnitStateData(Splitting, orphanData.dstBroker(), selectedBroker,
+ Map.copyOf(orphanData.splitServiceUnitToDestBroker()), true, version);
+ } else if (orphanData.state() == Owned) {
+ // if Owned, set orphan.dstBroker() as source to clean it up in case it is still
+ // alive.
+ return new ServiceUnitStateData(Owned, selectedBroker,
+ selectedBroker.equals(orphanData.dstBroker()) ? null :
+ orphanData.dstBroker(),
+ true, version);
+ } else {
+ // if Assigning or Releasing, set orphan.sourceBroker() as source
+ // to clean it up in case it is still alive.
+ return new ServiceUnitStateData(Owned, selectedBroker,
+ selectedBroker.equals(orphanData.sourceBroker()) ? null :
+ orphanData.sourceBroker(),
+ true, version);
+ }
+ // If no broker is selected(available), free the ownership.
+ // If the previous owner is still active, it will close the bundle(topic) ownership.
+ }).orElseGet(() -> new ServiceUnitStateData(Free, null,
+ orphanData.state() == Owned ? orphanData.dstBroker() : orphanData.sourceBroker(),
+ true,
+ version)))
+ .thenCompose(override -> {
+ log.info(
+ "Overriding inactiveBroker:{}, ownership serviceUnit:{} from orphanData:{} to "
+ + "overrideData:{}",
+ inactiveBroker, serviceUnit, orphanData, override);
+ return publishOverrideEventAsync(serviceUnit, override);
+ }).get(config.getMetadataStoreOperationTimeoutSeconds(), SECONDS);
+ } catch (Throwable e) {
+ log.error(
+ "Failed to override inactiveBroker:{} ownership serviceUnit:{} orphanData:{}. "
+ "totalCleanupErrorCnt:{}",
- serviceUnit, orphanData, totalCleanupErrorCnt.incrementAndGet());
- }
- var override = getOverrideInactiveBrokerStateData(orphanData, selectedBroker, inactiveBroker);
- log.info("Overriding ownership serviceUnit:{} from orphanData:{} to overrideData:{}",
- serviceUnit, orphanData, override);
- publishOverrideEventAsync(serviceUnit, orphanData, override)
- .exceptionally(e -> {
- log.error(
- "Failed to override the ownership serviceUnit:{} orphanData:{}. "
- + "Failed to publish override event. totalCleanupErrorCnt:{}",
- serviceUnit, orphanData, totalCleanupErrorCnt.incrementAndGet());
- return null;
- });
+ inactiveBroker, serviceUnit, orphanData, totalCleanupErrorCnt.incrementAndGet(), e);
+ }
}
private void waitForCleanups(String broker, boolean excludeSystemTopics, int maxWaitTimeInMillis) {
@@ -1386,60 +1385,13 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
- private Optional<String> selectBroker(String serviceUnit, String inactiveBroker) {
- try {
- return loadManager.selectAsync(
- LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit), Set.of(inactiveBroker))
- .get(inFlightStateWaitingTimeInMillis, MILLISECONDS);
- } catch (Throwable e) {
- log.error("Failed to select a broker for serviceUnit:{}", serviceUnit);
- }
- return Optional.empty();
- }
-
- private Optional<ServiceUnitStateData> getRollForwardStateData(String serviceUnit,
- String inactiveBroker,
- long nextVersionId) {
- Optional<String> selectedBroker = selectBroker(serviceUnit, inactiveBroker);
- if (selectedBroker.isEmpty()) {
- return Optional.empty();
- }
- return Optional.of(new ServiceUnitStateData(Owned, selectedBroker.get(), true, nextVersionId));
+ private CompletableFuture<Optional<String>> selectBroker(String serviceUnit, String inactiveBroker) {
+ return getLoadManager().selectAsync(
+ LoadManagerShared.getNamespaceBundle(pulsar, serviceUnit),
+ inactiveBroker == null ? Set.of() : Set.of(inactiveBroker));
}
- private Optional<ServiceUnitStateData> getOverrideInFlightStateData(
- String serviceUnit, ServiceUnitStateData orphanData,
- Set<String> availableBrokers) {
- long nextVersionId = getNextVersionId(orphanData);
- var state = orphanData.state();
- switch (state) {
- case Assigning: {
- return getRollForwardStateData(serviceUnit, orphanData.dstBroker(), nextVersionId);
- }
- case Splitting: {
- return Optional.of(new ServiceUnitStateData(Splitting,
- orphanData.dstBroker(), orphanData.sourceBroker(),
- Map.copyOf(orphanData.splitServiceUnitToDestBroker()),
- true, nextVersionId));
- }
- case Releasing: {
- if (availableBrokers.contains(orphanData.sourceBroker())) {
- // rollback to the src
- return Optional.of(new ServiceUnitStateData(Owned, orphanData.sourceBroker(), true, nextVersionId));
- } else {
- return getRollForwardStateData(serviceUnit, orphanData.sourceBroker(), nextVersionId);
- }
- }
- default: {
- var msg = String.format("Failed to get the overrideStateData from serviceUnit=%s, orphanData=%s",
- serviceUnit, orphanData);
- log.error(msg);
- throw new IllegalStateException(msg);
- }
- }
- }
-
@VisibleForTesting
protected void monitorOwnerships(List<String> brokers) {
if (!isChannelOwner()) {
@@ -1466,7 +1418,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
long startTime = System.nanoTime();
Set<String> inactiveBrokers = new HashSet<>();
Set<String> activeBrokers = new HashSet<>(brokers);
- Map<String, ServiceUnitStateData> orphanServiceUnits = new HashMap<>();
+ Map<String, ServiceUnitStateData> timedOutInFlightStateServiceUnits = new HashMap<>();
int serviceUnitTombstoneCleanupCnt = 0;
int orphanServiceUnitCleanupCnt = 0;
long totalCleanupErrorCntStart = totalCleanupErrorCnt.get();
@@ -1478,20 +1430,27 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
String srcBroker = stateData.sourceBroker();
var state = stateData.state();
- if (isActiveState(state) && StringUtils.isNotBlank(srcBroker) && !activeBrokers.contains(srcBroker)) {
+ if (state == Owned && (StringUtils.isBlank(dstBroker) || !activeBrokers.contains(dstBroker))) {
+ inactiveBrokers.add(dstBroker);
+ continue;
+ }
+
+ if (isInFlightState(state) && StringUtils.isNotBlank(srcBroker) && !activeBrokers.contains(srcBroker)) {
inactiveBrokers.add(srcBroker);
continue;
}
- if (isActiveState(state) && StringUtils.isNotBlank(dstBroker) && !activeBrokers.contains(dstBroker)) {
+ if (isInFlightState(state) && StringUtils.isNotBlank(dstBroker) && !activeBrokers.contains(dstBroker)) {
inactiveBrokers.add(dstBroker);
continue;
}
- if (isActiveState(state) && isInFlightState(state)
+
+ if (isInFlightState(state)
&& now - stateData.timestamp() > inFlightStateWaitingTimeInMillis) {
- orphanServiceUnits.put(serviceUnit, stateData);
+ timedOutInFlightStateServiceUnits.put(serviceUnit, stateData);
continue;
}
+
if (!isActiveState(state) && now - stateData.timestamp() > semiTerminalStateWaitingTimeInMillis) {
log.info("Found semi-terminal states to tombstone"
+ " serviceUnit:{}, stateData:{}", serviceUnit, stateData);
@@ -1507,37 +1466,21 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
}
}
- // Skip cleaning orphan bundles if inactiveBrokers exist. This is a bigger problem.
+
if (!inactiveBrokers.isEmpty()) {
for (String inactiveBroker : inactiveBrokers) {
handleBrokerDeletionEvent(inactiveBroker);
}
- } else if (!orphanServiceUnits.isEmpty()) {
- for (var etr : orphanServiceUnits.entrySet()) {
+ }
+
+ // timedOutInFlightStateServiceUnits are the in-flight ones although their src and dst brokers are known to
+ // be active.
+ if (!timedOutInFlightStateServiceUnits.isEmpty()) {
+ for (var etr : timedOutInFlightStateServiceUnits.entrySet()) {
var orphanServiceUnit = etr.getKey();
var orphanData = etr.getValue();
- var overrideData = getOverrideInFlightStateData(
- orphanServiceUnit, orphanData, activeBrokers);
- if (overrideData.isPresent()) {
- log.info("Overriding in-flight state ownership serviceUnit:{} "
- + "from orphanData:{} to overrideData:{}",
- orphanServiceUnit, orphanData, overrideData);
- publishOverrideEventAsync(orphanServiceUnit, orphanData, overrideData.get())
- .whenComplete((__, e) -> {
- if (e != null) {
- log.error("Failed cleaning the ownership orphanServiceUnit:{}, orphanData:{}, "
- + "cleanupErrorCnt:{}.",
- orphanServiceUnit, orphanData,
- totalCleanupErrorCnt.incrementAndGet() - totalCleanupErrorCntStart, e);
- }
- });
- orphanServiceUnitCleanupCnt++;
- } else {
- log.warn("Failed get the overrideStateData from orphanServiceUnit:{}, orphanData:{},"
- + " cleanupErrorCnt:{}. will retry..",
- orphanServiceUnit, orphanData,
- totalCleanupErrorCnt.incrementAndGet() - totalCleanupErrorCntStart);
- }
+ overrideOwnership(orphanServiceUnit, orphanData, null);
+ orphanServiceUnitCleanupCnt++;
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 9b11df6aeb9..18ad0fba9cd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -2247,6 +2247,15 @@ public class BrokerService implements Closeable {
topics.forEach((name, topicFuture) -> {
TopicName topicName = TopicName.get(name);
if (serviceUnit.includes(topicName)) {
+ if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)
+ && ExtensibleLoadManagerImpl.isInternalTopic(topicName.toString())) {
+ if (ExtensibleLoadManagerImpl.debug(pulsar.getConfiguration(), log)) {
+ log.info("[{}] Skip unloading ExtensibleLoadManager internal topics. Such internal topic "
+ + "should be closed when shutting down the broker.", topicName);
+ }
+ return;
+ }
+
// Topic needs to be unloaded
log.info("[{}] Unloading topic", topicName);
if (topicFuture.isCompletedExceptionally()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
index ceb58e8d964..fc34b8b5990 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java
@@ -85,6 +85,7 @@ import org.apache.pulsar.broker.loadbalance.extensions.models.Unload;
import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
+import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.client.impl.TableViewImpl;
@@ -108,6 +109,7 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
private ServiceUnitStateChannel channel2;
private String brokerId1;
private String brokerId2;
+ private String brokerId3;
private String bundle;
private String bundle1;
private String bundle2;
@@ -159,6 +161,7 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
FieldUtils.readDeclaredField(channel1, "brokerId", true);
brokerId2 = (String)
FieldUtils.readDeclaredField(channel2, "brokerId", true);
+ brokerId3 = "broker-3";
bundle = "public/default/0x00000000_0xffffffff";
bundle1 = "public/default/0x00000000_0xfffffff0";
@@ -172,6 +175,19 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
childBundle31 = "public/default3/" + childBundle1Range;
childBundle32 = "public/default3/" + childBundle2Range;
+
+ var testNamespaces = List.of("public/releasing1", "public/releasing2", "public/assigning1", "public/assigning2",
+ "public/free", "public/deleted", "public/owned1", "public/owned2SourceBundle", "public/owned3");
+
+ for (var ns : testNamespaces) {
+ admin.namespaces().createNamespace(ns);
+ }
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
+ var actual = admin.namespaces().getNamespaces("public");
+ System.out.println(actual);
+ assertTrue(actual.containsAll(testNamespaces));
+ }
+ );
}
@BeforeMethod
@@ -183,6 +199,8 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
cleanOpsCounters(channel2);
cleanMetadataState(channel1);
cleanMetadataState(channel2);
+ getCleanupJobs(channel1).clear();
+ getCleanupJobs(channel2).clear();
}
@@ -1423,6 +1441,8 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
assertEquals(3, count.get());
});
var leader = channel1.isChannelOwnerAsync().get() ? channel1 : channel2;
+ doReturn(CompletableFuture.completedFuture(Optional.of(brokerId1)))
+ .when(loadManager).selectAsync(any(), any());
((ServiceUnitStateChannelImpl) leader)
.monitorOwnerships(List.of(brokerId1, brokerId2));
waitUntilState(leader, bundle3, Deleted);
@@ -1549,7 +1569,8 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
@Test(priority = 18)
public void testOverrideOrphanStateData()
- throws IllegalAccessException, ExecutionException, InterruptedException, TimeoutException {
+ throws IllegalAccessException, ExecutionException, InterruptedException, TimeoutException,
+ PulsarAdminException {
var leaderChannel = channel1;
var followerChannel = channel2;
@@ -1564,26 +1585,41 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
String broker = brokerId1;
// test override states
- String releasingBundle = "public/releasing/0xfffffff0_0xffffffff";
+ String releasingBundle1 = "public/releasing1/0xfffffff0_0xffffffff";
+ String releasingBundle2 = "public/releasing2/0xfffffff0_0xffffffff";
String splittingBundle = bundle;
- String assigningBundle = "public/assigning/0xfffffff0_0xffffffff";
+ String assigningBundle1 = "public/assigning1/0xfffffff0_0xffffffff";
+ String assigningBundle2 = "public/assigning2/0xfffffff0_0xffffffff";
String freeBundle = "public/free/0xfffffff0_0xffffffff";
String deletedBundle = "public/deleted/0xfffffff0_0xffffffff";
- String ownedBundle = "public/owned/0xfffffff0_0xffffffff";
- overrideTableViews(releasingBundle,
- new ServiceUnitStateData(Releasing, null, broker, 1));
+ String ownedBundle1 = "public/owned1/0xfffffff0_0xffffffff";
+ String ownedBundle2 = "public/owned2SourceBundle/0xfffffff0_0xffffffff";
+ String ownedBundle3 = "public/owned3/0xfffffff0_0xffffffff";
+ String inactiveBroker = "broker-inactive-1";
+
+ overrideTableViews(releasingBundle1,
+ new ServiceUnitStateData(Releasing, broker, brokerId2, 1));
+ overrideTableViews(releasingBundle2,
+ new ServiceUnitStateData(Releasing, brokerId2, brokerId3, 1));
overrideTableViews(splittingBundle,
new ServiceUnitStateData(Splitting, null, broker,
Map.of(childBundle1Range, Optional.empty(),
childBundle2Range, Optional.empty()), 1));
- overrideTableViews(assigningBundle,
+ overrideTableViews(assigningBundle1,
new ServiceUnitStateData(Assigning, broker, null, 1));
+ overrideTableViews(assigningBundle2,
+ new ServiceUnitStateData(Assigning, broker, brokerId2, 1));
overrideTableViews(freeBundle,
new ServiceUnitStateData(Free, null, broker, 1));
overrideTableViews(deletedBundle,
new ServiceUnitStateData(Deleted, null, broker, 1));
- overrideTableViews(ownedBundle,
+ overrideTableViews(ownedBundle1,
new ServiceUnitStateData(Owned, broker, null, 1));
+ overrideTableViews(ownedBundle2,
+ new ServiceUnitStateData(Owned, broker, inactiveBroker, 1));
+ overrideTableViews(ownedBundle3,
+ new ServiceUnitStateData(Owned, inactiveBroker, broker, 1));
+
// test stable metadata state
doReturn(CompletableFuture.completedFuture(Optional.of(brokerId2)))
@@ -1593,16 +1629,21 @@ public class ServiceUnitStateChannelTest extends MockedPulsarServiceBaseTest {
FieldUtils.writeDeclaredField(followerChannel, "inFlightStateWaitingTimeInMillis",
-1, true);
((ServiceUnitStateChannelImpl) leaderChannel)
- .monitorOwnerships(List.of(brokerId1, brokerId2));
+ .monitorOwnerships(List.of(brokerId1, brokerId2, "broker-3"));
- waitUntilNewOwner(channel2, releasingBundle, broker);
- waitUntilNewOwner(channel2, childBundle11, broker);
- waitUntilNewOwner(channel2, childBundle12, broker);
- waitUntilNewOwner(channel2, assigningBundle, brokerId2);
- waitUntilNewOwner(channel2, ownedBundle, broker);
- assertEquals(Optional.empty(), channel2.getOwnerAsync(freeBundle).get());
- assertTrue(channel2.getOwnerAsync(deletedBundle).isCompletedExceptionally());
+
+ waitUntilNewOwner(channel2, releasingBundle1, brokerId2);
+ waitUntilNewOwner(channel2, releasingBundle2, brokerId2);
assertTrue(channel2.getOwnerAsync(splittingBundle).isCompletedExceptionally());
+ waitUntilNewOwner(channel2, childBundle11, brokerId2);
+ waitUntilNewOwner(channel2, childBundle12, brokerId2);
+ waitUntilNewOwner(channel2, assigningBundle1, brokerId2);
+ waitUntilNewOwner(channel2, assigningBundle2, brokerId2);
+ assertTrue(channel2.getOwnerAsync(freeBundle).get().isEmpty());
+ assertTrue(channel2.getOwnerAsync(deletedBundle).isCompletedExceptionally());
+ waitUntilNewOwner(channel2, ownedBundle1, broker);
+ waitUntilNewOwner(channel2, ownedBundle2, broker);
+ waitUntilNewOwner(channel2, ownedBundle3, brokerId2);
// clean-up
FieldUtils.writeDeclaredField(channel1,