From d0963b64453d380c2b257c5f9587b465315faf1e Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 24 Apr 2024 23:21:47 +0800 Subject: [PATCH] [fix] [broker] [namespace Geo-Replication]Reject a topic creation if there is a confilct topic on the remote side --- .../pulsar/broker/admin/AdminResource.java | 19 +++--- .../broker/service/OneWayReplicatorTest.java | 59 +++++++++++++++++++ .../service/OneWayReplicatorTestBase.java | 17 +++++- .../apache/pulsar/common/util/FutureUtil.java | 3 + 4 files changed, 88 insertions(+), 10 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index 45455f16d4dc1..b7172f2f7d92e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -605,12 +605,15 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n throw new RestException(Status.CONFLICT, "This topic already exists"); } }) + .thenCompose(__ -> { + if (!createLocalTopicOnly && topicName.isGlobal()) { + return internalCreatePartitionedTopicToReplicatedClustersInBackground(numPartitions); + } + return CompletableFuture.completedFuture(null); + }) .thenCompose(__ -> provisionPartitionedTopicPath(numPartitions, createLocalTopicOnly, properties)) .thenCompose(__ -> tryCreatePartitionsAsync(numPartitions)) .thenRun(() -> { - if (!createLocalTopicOnly && topicName.isGlobal()) { - internalCreatePartitionedTopicToReplicatedClustersInBackground(numPartitions); - } log.info("[{}] Successfully created partitions for topic {} in cluster {}", clientAppId(), topicName, pulsar().getConfiguration().getClusterName()); asyncResponse.resume(Response.noContent().build()); @@ -622,11 +625,13 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n }); } - private void internalCreatePartitionedTopicToReplicatedClustersInBackground(int numPartitions) { - getNamespaceReplicatedClustersAsync(namespaceName) - .thenAccept(clusters -> { + private CompletableFuture internalCreatePartitionedTopicToReplicatedClustersInBackground(int numPartitions) { + return getNamespaceReplicatedClustersAsync(namespaceName) + .thenCompose(clusters -> { // this call happens in the background without async composition. completion is logged. - internalCreatePartitionedTopicToReplicatedClustersInBackground(clusters, numPartitions); + Map> createPartitionedMetaOnRemoteCluster = + internalCreatePartitionedTopicToReplicatedClustersInBackground(clusters, numPartitions); + return FutureUtil.waitForAll(createPartitionedMetaOnRemoteCluster.values()); }); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java index eb31c13b0d528..7ad404b5f2953 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java @@ -663,4 +663,63 @@ public void testUnFenceTopicToReuse() throws Exception { admin2.topics().delete(topicName); }); } + + @Test + public void testNamespaceLevelReplicationRemoteConflictTopicExist() throws Exception { + final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp"); + // Verify: will get a not found error when calling "getPartitionedTopicMetadata" on a topic not exists. + try { + admin1.topics().getPartitionedTopicMetadata(topicName); + fail("Expected a not found error"); + } catch (Exception ex) { + Throwable unWrapEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(unWrapEx.getMessage().contains("not found")); + } + // Verify: will get a conflict error when there is a topic with different partitions on the remote side. + admin2.topics().createPartitionedTopic(topicName, 1); + try { + admin1.topics().createPartitionedTopic(topicName, 2); + fail("Expected error due to a conflict partitioned topic already exists."); + } catch (Exception ex) { + Throwable unWrapEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(unWrapEx.getMessage().contains("with different partitions")); + } + // Verify: nothing has been changed after the failed calling. + Optional partitions1 = pulsar1.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(TopicName.get(topicName), true).join(); + assertFalse(partitions1.isPresent()); + Optional partitions2 = pulsar2.getPulsarResources().getNamespaceResources() + .getPartitionedTopicResources() + .getPartitionedTopicMetadataAsync(TopicName.get(topicName), true).join(); + assertTrue(partitions2.isPresent()); + assertEquals(partitions2.get().partitions, 1); + // cleanup. + admin2.topics().deletePartitionedTopic(topicName); + } + + @Test + public void testNamespaceLevelPartitionedMetadataReplication() throws Exception { + final String topicName = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp"); + // Verify: will get a not found error when calling "getPartitionedTopicMetadata" on a topic not exists. + try { + admin1.topics().getPartitionedTopicMetadata(topicName); + fail("Expected a not found error"); + } catch (Exception ex) { + Throwable unWrapEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(unWrapEx.getMessage().contains("not found")); + } + // Verify: will get a conflict error when there is a topic with different partitions on the remote side. + admin1.topics().createPartitionedTopic(topicName, 2); + // Verify: nothing has been changed after the failed calling. + PartitionedTopicMetadata topicMetadata1 = admin1.topics().getPartitionedTopicMetadata(topicName); + assertEquals(topicMetadata1.partitions, 2); + PartitionedTopicMetadata topicMetadata2 = admin2.topics().getPartitionedTopicMetadata(topicName); + assertEquals(topicMetadata2.partitions, 2); + // cleanup. + cleanupTopics(() -> { + admin1.topics().deletePartitionedTopic(topicName); + admin2.topics().deletePartitionedTopic(topicName); + }); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java index b4eed00c4470f..afe1117069fda 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java @@ -24,6 +24,7 @@ import java.time.Duration; import java.util.Collections; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; @@ -163,9 +164,19 @@ protected void cleanupTopics(String namespace, CleanupTopicAction cleanupTopicAc } protected void waitChangeEventsInit(String namespace) { - PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService() - .getTopic(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, false) - .join().get(); + if (!pulsar1.getConfig().isSystemTopicAndTopicLevelPoliciesEnabled() || pulsar1.getConfig().isSystemTopicEnabled()) { + return; + } + CompletableFuture> future = pulsar1.getBrokerService() + .getTopic(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, false); + if (future == null) { + return; + } + Optional optional = future.join(); + if (optional.isEmpty()) { + return; + } + PersistentTopic topic = (PersistentTopic) optional.get(); Awaitility.await().atMost(Duration.ofSeconds(180)).untilAsserted(() -> { TopicStatsImpl topicStats = topic.getStats(true, false, false); topicStats.getSubscriptions().entrySet().forEach(entry -> { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java index f6fcb12f35939..12724ff587b44 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java @@ -54,6 +54,9 @@ public class FutureUtil { * @return a new CompletableFuture that is completed when all of the given CompletableFutures complete */ public static CompletableFuture waitForAll(Collection> futures) { + if (futures.isEmpty()) { + return CompletableFuture.completedFuture(null); + } return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); }