Revert Add listener interface for namespace service #20406
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 717ff48..4eb2584 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
@@ -930,7 +930,6 @@
double splitBundleTime = TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - startTime));
log.info("Successfully split {} parent namespace-bundle to {} in {} ms",
parentBundle, childBundles, splitBundleTime);
- namespaceService.onNamespaceBundleSplit(parentBundle);
completionFuture.complete(null);
})
.exceptionally(ex -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceBundleSplitListener.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceBundleSplitListener.java
deleted file mode 100644
index a3312f5..0000000
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceBundleSplitListener.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.broker.namespace;
-
-import java.util.function.Predicate;
-import org.apache.pulsar.common.naming.NamespaceBundle;
-
-/**
- * Listener for <code>NamespaceBundle</code> split.
- */
-public interface NamespaceBundleSplitListener extends Predicate<NamespaceBundle> {
- void onSplit(NamespaceBundle bundle);
-}
\ No newline at end of file
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 9be8d49..141e485 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -139,10 +139,6 @@
private final ConcurrentOpenHashMap<ClusterDataImpl, PulsarClientImpl> namespaceClients;
private final List<NamespaceBundleOwnershipListener> bundleOwnershipListeners;
-
- private final List<NamespaceBundleSplitListener> bundleSplitListeners;
-
-
private final RedirectManager redirectManager;
@@ -171,7 +167,6 @@
this.namespaceClients =
ConcurrentOpenHashMap.<ClusterDataImpl, PulsarClientImpl>newBuilder().build();
this.bundleOwnershipListeners = new CopyOnWriteArrayList<>();
- this.bundleSplitListeners = new CopyOnWriteArrayList<>();
this.localBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(LocalBrokerData.class);
this.redirectManager = new RedirectManager(pulsar);
}
@@ -1018,7 +1013,6 @@
// affect the split operation which is already safely completed
r.forEach(this::unloadNamespaceBundle);
}
- onNamespaceBundleSplit(bundle);
})
.exceptionally(e -> {
String msg1 = format(
@@ -1261,19 +1255,6 @@
}
}
}
-
- public void onNamespaceBundleSplit(NamespaceBundle bundle) {
- for (NamespaceBundleSplitListener bundleSplitListener : bundleSplitListeners) {
- try {
- if (bundleSplitListener.test(bundle)) {
- bundleSplitListener.onSplit(bundle);
- }
- } catch (Throwable t) {
- LOG.error("Call bundle {} split listener {} error", bundle, bundleSplitListener, t);
- }
- }
- }
-
public void addNamespaceBundleOwnershipListener(NamespaceBundleOwnershipListener... listeners) {
Objects.requireNonNull(listeners);
for (NamespaceBundleOwnershipListener listener : listeners) {
@@ -1284,15 +1265,6 @@
getOwnedServiceUnits().forEach(bundle -> notifyNamespaceBundleOwnershipListener(bundle, listeners));
}
- public void addNamespaceBundleSplitListener(NamespaceBundleSplitListener... listeners) {
- Objects.requireNonNull(listeners);
- for (NamespaceBundleSplitListener listener : listeners) {
- if (listener != null) {
- bundleSplitListeners.add(listener);
- }
- }
- }
-
private void notifyNamespaceBundleOwnershipListener(NamespaceBundle bundle,
NamespaceBundleOwnershipListener... listeners) {
if (listeners != null) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
index 533e617..cff92ab 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplTest.java
@@ -88,7 +88,6 @@
import org.apache.pulsar.broker.lookup.LookupResult;
import org.apache.pulsar.broker.namespace.LookupOptions;
import org.apache.pulsar.broker.namespace.NamespaceBundleOwnershipListener;
-import org.apache.pulsar.broker.namespace.NamespaceBundleSplitListener;
import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
@@ -406,23 +405,6 @@
String firstBundle = bundleRanges.get(0) + "_" + bundleRanges.get(1);
- AtomicInteger splitCount = new AtomicInteger(0);
- NamespaceBundleSplitListener namespaceBundleSplitListener = new NamespaceBundleSplitListener() {
- @Override
- public void onSplit(NamespaceBundle bundle) {
- splitCount.incrementAndGet();
- }
-
- @Override
- public boolean test(NamespaceBundle namespaceBundle) {
- return namespaceBundle
- .toString()
- .equals(String.format(namespace + "/0x%08x_0x%08x", bundleRanges.get(0), bundleRanges.get(1)));
- }
- };
- pulsar1.getNamespaceService().addNamespaceBundleSplitListener(namespaceBundleSplitListener);
- pulsar2.getNamespaceService().addNamespaceBundleSplitListener(namespaceBundleSplitListener);
-
long mid = bundleRanges.get(0) + (bundleRanges.get(1) - bundleRanges.get(0)) / 2;
admin.namespaces().splitNamespaceBundle(namespace, firstBundle, true, null);
@@ -435,7 +417,6 @@
assertTrue(bundlesData.getBoundaries().contains(lowBundle));
assertTrue(bundlesData.getBoundaries().contains(midBundle));
assertTrue(bundlesData.getBoundaries().contains(highBundle));
- assertEquals(splitCount.get(), 1);
// Test split bundle with invalid bundle range.
try {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java
index 73cfaf1..43d3746 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceCreateBundlesTest.java
@@ -20,21 +20,15 @@
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
-
-import lombok.Cleanup;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
-import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.policies.data.BookieAffinityGroupData;
import org.apache.pulsar.common.policies.data.Policies;
-import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -87,31 +81,4 @@
assertNotNull(admin.namespaces().getBookieAffinityGroup(namespaceName));
producer.close();
}
-
- @Test
- public void testBundleSplitListener() throws Exception {
- String namespaceName = "prop/" + UUID.randomUUID().toString();
- String topicName = "persistent://" + namespaceName + "/my-topic5";
- admin.namespaces().createNamespace(namespaceName);
- @Cleanup
- Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).sendTimeout(1,
- TimeUnit.SECONDS).create();
- producer.send(new byte[1]);
- String bundleRange = admin.lookups().getBundleRange(topicName);
- AtomicBoolean isTriggered = new AtomicBoolean(false);
- pulsar.getNamespaceService().addNamespaceBundleSplitListener(new NamespaceBundleSplitListener() {
- @Override
- public void onSplit(NamespaceBundle bundle) {
- assertEquals(bundleRange, bundle.getBundleRange());
- isTriggered.set(true);
- }
-
- @Override
- public boolean test(NamespaceBundle namespaceBundle) {
- return true;
- }
- });
- admin.namespaces().splitNamespaceBundle(namespaceName, bundleRange, false, null);
- Awaitility.await().untilAsserted(() -> assertTrue(isTriggered.get()));
- }
}
