From cf0da5e5b5aa4886d74582642b9845f88dcbfac1 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 25 Nov 2025 11:34:13 +0800 Subject: [PATCH 1/6] [improve][broker] PIP-433: Ensure topic creation before starting GEO --- .../broker/service/AbstractReplicator.java | 6 +- .../NonPersistentReplicator.java | 4 +- .../nonpersistent/NonPersistentTopic.java | 13 +- .../persistent/GeoPersistentReplicator.java | 121 ++++++++++- .../persistent/PersistentReplicator.java | 8 +- .../service/persistent/PersistentTopic.java | 23 ++- .../service/persistent/ShadowReplicator.java | 6 +- .../service/AbstractReplicatorTest.java | 21 +- .../broker/service/OneWayReplicatorTest.java | 193 ++++++++++++++++++ .../broker/service/PersistentTopicTest.java | 44 +++- 10 files changed, 405 insertions(+), 34 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java index f996d328090ca..6975ba86e8d90 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java @@ -33,6 +33,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.NamingException; import org.apache.pulsar.broker.service.BrokerServiceException.TopicBusyException; import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.MessageRoutingMode; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; @@ -57,6 +58,7 @@ public abstract class AbstractReplicator implements Replicator { protected final String remoteCluster; protected final PulsarClientImpl replicationClient; protected final PulsarClientImpl client; + protected final PulsarAdmin replicationAdmin; protected String replicatorId; @Getter protected final Topic localTopic; @@ -107,7 +109,8 @@ public enum State { } public AbstractReplicator(String localCluster, Topic localTopic, String remoteCluster, String remoteTopicName, - String replicatorPrefix, BrokerService brokerService, PulsarClientImpl replicationClient) + String replicatorPrefix, BrokerService brokerService, PulsarClientImpl replicationClient, + PulsarAdmin replicationAdmin) throws PulsarServerException { this.brokerService = brokerService; this.localTopic = localTopic; @@ -117,6 +120,7 @@ public AbstractReplicator(String localCluster, Topic localTopic, String remoteCl this.remoteTopicName = remoteTopicName; this.remoteCluster = StringInterner.intern(remoteCluster); this.replicationClient = replicationClient; + this.replicationAdmin = replicationAdmin; this.client = (PulsarClientImpl) brokerService.pulsar().getClient(); this.producer = null; this.producerQueueSize = brokerService.pulsar().getConfiguration().getReplicationProducerQueueSize(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java index 38e1894c17854..076c0aa29353d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java @@ -30,6 +30,7 @@ import org.apache.pulsar.broker.service.AbstractReplicator; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.Replicator; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; @@ -51,7 +52,8 @@ public class NonPersistentReplicator extends AbstractReplicator implements Repli private final NonPersistentReplicatorStatsImpl stats = new NonPersistentReplicatorStatsImpl(); public NonPersistentReplicator(NonPersistentTopic topic, String localCluster, String remoteCluster, - BrokerService brokerService, PulsarClientImpl replicationClient) throws PulsarServerException { + BrokerService brokerService, PulsarClientImpl replicationClient, + PulsarAdmin replicationAdmin) throws PulsarServerException { super(localCluster, topic, remoteCluster, topic.getName(), topic.getReplicatorPrefix(), brokerService, replicationClient); // NonPersistentReplicator does not support limitation so far, so reset pending queue size to the default value. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 6021c41142a5e..96a9f97d70f50 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -71,7 +71,9 @@ import org.apache.pulsar.broker.service.schema.exceptions.NotExistSchemaException; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.stats.NamespaceStats; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition; @@ -628,14 +630,15 @@ protected CompletableFuture addReplicationCluster(String remoteCluster, No String localCluster) { return AbstractReplicator.validatePartitionedTopicAsync(nonPersistentTopic.getName(), brokerService) .thenCompose(__ -> brokerService.pulsar().getPulsarResources().getClusterResources() - .getClusterAsync(remoteCluster) - .thenApply(clusterData -> - brokerService.getReplicationClient(remoteCluster, clusterData))) - .thenAccept(replicationClient -> { + .getClusterAsync(remoteCluster)) + .thenAccept((clusterData) -> { + PulsarClient replicationClient = brokerService.getReplicationClient(remoteCluster, clusterData); + PulsarAdmin replicationAdmin = brokerService.getClusterPulsarAdmin(remoteCluster, clusterData); replicators.computeIfAbsent(remoteCluster, r -> { try { return new NonPersistentReplicator(NonPersistentTopic.this, localCluster, - remoteCluster, brokerService, (PulsarClientImpl) replicationClient); + remoteCluster, brokerService, (PulsarClientImpl) replicationClient, + replicationAdmin); } catch (PulsarServerException e) { log.error("[{}] Replicator startup failed {}", topic, remoteCluster, e); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index 46f8a27d58020..9ee84f0c7ce8b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -22,15 +22,21 @@ import io.netty.buffer.ByteBuf; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException; +import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.protocol.Markers; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.util.FutureUtil; @@ -40,9 +46,10 @@ public class GeoPersistentReplicator extends PersistentReplicator { public GeoPersistentReplicator(PersistentTopic topic, ManagedCursor cursor, String localCluster, String remoteCluster, BrokerService brokerService, - PulsarClientImpl replicationClient) + PulsarClientImpl replicationClient, PulsarAdmin replicationAdmin) throws PulsarServerException { - super(localCluster, topic, cursor, remoteCluster, topic.getName(), brokerService, replicationClient); + super(localCluster, topic, cursor, remoteCluster, topic.getName(), brokerService, replicationClient, + replicationAdmin); } /** @@ -56,7 +63,115 @@ protected String getProducerName() { @Override protected CompletableFuture prepareCreateProducer() { if (brokerService.getPulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) { - return CompletableFuture.completedFuture(null); + TopicName completeTopicName = TopicName.get(localTopicName); + TopicName baseTopicName; + if (completeTopicName.isPartitioned()) { + baseTopicName = TopicName.get(completeTopicName.getPartitionedTopicName()); + } else { + baseTopicName = completeTopicName; + } + // Set useFallbackForNonPIP344Brokers to true when mix of PIP-344 and non-PIP-344 brokers are used, it + // can still work. + return client.getLookup().getPartitionedTopicMetadata(baseTopicName, false, true) + .thenCompose((localMetadata) -> replicationAdmin.topics() + // https://github.com/apache/pulsar/pull/4963 + // Use the admin API instead of the client to fetch partitioned metadata + // to prevent automatic topic creation on the remote cluster. + // PIP-344 introduced an option to disable auto-creation when fetching partitioned + // topic metadata via the client, but this requires Pulsar 3.0.x. + // This change is a workaround to support Pulsar 2.4.2. + .getPartitionedTopicMetadataAsync(baseTopicName.toString()) + .exceptionally(ex -> { + Throwable throwable = FutureUtil.unwrapCompletionException(ex); + if (throwable instanceof NotFoundException) { + // Topic does not exist on the remote cluster. + return new PartitionedTopicMetadata(0); + } + throw new CompletionException("Failed to get partitioned topic metadata", throwable); + }).thenCompose(remoteMetadata -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Local metadata partitions: {} Remote metadata partitions: {}", + replicatorId, localMetadata.partitions, remoteMetadata.partitions); + } + + // Non-partitioned topic + if (localMetadata.partitions == 0) { + if (localMetadata.partitions == remoteMetadata.partitions) { + return replicationAdmin.topics().createNonPartitionedTopicAsync(localTopicName) + .exceptionally(ex -> { + Throwable throwable = FutureUtil.unwrapCompletionException(ex); + if (throwable instanceof ConflictException) { + // Topic already exists on the remote cluster. + return null; + } else { + throw new CompletionException( + "Failed to create non-partitioned topic", throwable); + } + }); + } else { + return FutureUtil.failedFuture(new PulsarClientException.NotAllowedException( + "Topic type is not matched between local and remote cluster: local " + + "partitions: " + localMetadata.partitions + + ", remote partitions: " + remoteMetadata.partitions)); + } + } else { + if (remoteMetadata.partitions == 0) { + if (log.isDebugEnabled()) { + log.debug("[{}] Creating partitioned topic {} with {} partitions", + replicatorId, baseTopicName, localMetadata.partitions); + } + // We maybe need to create a partitioned topic on remote cluster. + return replicationAdmin.topics() + .createPartitionedTopicAsync(baseTopicName.toString(), + localMetadata.partitions) + .exceptionally(ex -> { + Throwable throwable = FutureUtil.unwrapCompletionException(ex); + if (throwable instanceof ConflictException) { + // Topic already exists on the remote cluster. + // This can happen if the topic was created, or the topic is + // non-partitioned. + return null; + } else { + throw new CompletionException( + "Failed to create partitioned topic", throwable); + } + }) + .thenCompose((__) -> replicationAdmin.topics() + .getPartitionedTopicMetadataAsync(baseTopicName.toString())) + .thenCompose(metadata -> { + // Double check if the partitioned topic is created + // successfully. + // When partitions is equals to 0, it means this topic is + // non-partitioned, we should throw an exception. + if (completeTopicName.getPartitionIndex() >= metadata.partitions) { + return FutureUtil.failedFuture( + new PulsarClientException.NotAllowedException( + "Topic type is not matched between " + + "local and " + + "remote cluster: local " + + "partitions: " + + localMetadata.partitions + + ", remote partitions: " + + remoteMetadata.partitions)); + } + return CompletableFuture.completedFuture(null); + }); + } else { + if (localMetadata.partitions != remoteMetadata.partitions) { + return FutureUtil.failedFuture( + new PulsarClientException.NotAllowedException( + "The number of topic partitions is inconsistent between " + + "local and" + + " remote " + + "clusters: local partitions: " + + localMetadata.partitions + + ", remote partitions: " + + remoteMetadata.partitions)); + } + } + } + return CompletableFuture.completedFuture(null); + })); } else { CompletableFuture topicCheckFuture = new CompletableFuture<>(); replicationClient.getPartitionedTopicMetadata(localTopic.getName(), false, false) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java index e0a31476fc9f7..c1d73cd389183 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java @@ -58,6 +58,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.MessageExpirer; import org.apache.pulsar.broker.service.Replicator; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; @@ -119,11 +120,12 @@ protected enum ReasonOfWaitForCursorRewinding { protected final LinkedList inFlightTasks = new LinkedList<>(); public PersistentReplicator(String localCluster, PersistentTopic localTopic, ManagedCursor cursor, - String remoteCluster, String remoteTopic, - BrokerService brokerService, PulsarClientImpl replicationClient) + String remoteCluster, String remoteTopic, + BrokerService brokerService, PulsarClientImpl replicationClient, + PulsarAdmin replicationAdmin) throws PulsarServerException { super(localCluster, localTopic, remoteCluster, remoteTopic, localTopic.getReplicatorPrefix(), - brokerService, replicationClient); + brokerService, replicationClient, replicationAdmin); this.topic = localTopic; this.localSchemaTopicName = TopicName.getPartitionedTopicName(localTopicName).toString(); this.cursor = Objects.requireNonNull(cursor); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 451471e215ea5..5a1e0e940a802 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -148,6 +148,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException; import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.BatchMessageIdImpl; import org.apache.pulsar.client.impl.MessageIdImpl; @@ -2343,11 +2344,11 @@ protected CompletableFuture addReplicationCluster(String remoteCluster, Ma String localCluster) { return AbstractReplicator.validatePartitionedTopicAsync(PersistentTopic.this.getName(), brokerService) .thenCompose(__ -> brokerService.pulsar().getPulsarResources().getClusterResources() - .getClusterAsync(remoteCluster) - .thenApply(clusterData -> - brokerService.getReplicationClient(remoteCluster, clusterData))) - .thenAccept(replicationClient -> { - if (replicationClient == null) { + .getClusterAsync(remoteCluster)) + .thenAccept((clusterData) -> { + PulsarClient replicationClient = brokerService.getReplicationClient(remoteCluster, clusterData); + PulsarAdmin replicationAdmin = brokerService.getClusterPulsarAdmin(remoteCluster, clusterData); + if (replicationClient == null || replicationAdmin == null) { log.error("[{}] Can not create replicator because the remote client can not be created." + " remote cluster: {}. State of transferring : {}", topic, remoteCluster, transferring); @@ -2365,7 +2366,8 @@ protected CompletableFuture addReplicationCluster(String remoteCluster, Ma Replicator replicator = replicators.computeIfAbsent(remoteCluster, r -> { try { return new GeoPersistentReplicator(PersistentTopic.this, cursor, localCluster, - remoteCluster, brokerService, (PulsarClientImpl) replicationClient); + remoteCluster, brokerService, (PulsarClientImpl) replicationClient, + replicationAdmin); } catch (PulsarServerException e) { log.error("[{}] Replicator startup failed {}", topic, remoteCluster, e); } @@ -2431,9 +2433,10 @@ protected CompletableFuture addShadowReplicationCluster(String shadowTopic String localCluster = brokerService.pulsar().getConfiguration().getClusterName(); return AbstractReplicator.validatePartitionedTopicAsync(PersistentTopic.this.getName(), brokerService) .thenCompose(__ -> brokerService.pulsar().getPulsarResources().getClusterResources() - .getClusterAsync(localCluster) - .thenApply(clusterData -> brokerService.getReplicationClient(localCluster, clusterData))) - .thenAccept(replicationClient -> { + .getClusterAsync(localCluster)) + .thenAccept((clusterData) -> { + PulsarClient replicationClient = brokerService.getReplicationClient(localCluster, clusterData); + PulsarAdmin replicationAdmin = brokerService.getClusterPulsarAdmin(localCluster, clusterData); Replicator replicator = shadowReplicators.computeIfAbsent(shadowTopic, r -> { try { TopicName sourceTopicName = TopicName.get(getName()); @@ -2442,7 +2445,7 @@ protected CompletableFuture addShadowReplicationCluster(String shadowTopic shadowPartitionTopic += "-partition-" + sourceTopicName.getPartitionIndex(); } return new ShadowReplicator(shadowPartitionTopic, PersistentTopic.this, cursor, - brokerService, (PulsarClientImpl) replicationClient); + brokerService, (PulsarClientImpl) replicationClient, replicationAdmin); } catch (PulsarServerException e) { log.error("[{}] ShadowReplicator startup failed {}", topic, shadowTopic, e); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java index a334fd86dd02f..2e5e91fb9a633 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java @@ -27,6 +27,7 @@ import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; @@ -39,11 +40,12 @@ public class ShadowReplicator extends PersistentReplicator { public ShadowReplicator(String shadowTopic, PersistentTopic sourceTopic, ManagedCursor cursor, - BrokerService brokerService, PulsarClientImpl replicationClient) + BrokerService brokerService, PulsarClientImpl replicationClient, + PulsarAdmin replicationAdmin) throws PulsarServerException { super(brokerService.pulsar().getConfiguration().getClusterName(), sourceTopic, cursor, brokerService.pulsar().getConfiguration().getClusterName(), shadowTopic, brokerService, - replicationClient); + replicationClient, replicationAdmin); } /** diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java index 39252ac0a94e6..5f1d3a8a6c50b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/AbstractReplicatorTest.java @@ -22,6 +22,7 @@ import static org.mockito.Mockito.anyBoolean; import static org.mockito.Mockito.anyInt; import static org.mockito.Mockito.anyString; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import io.netty.channel.DefaultEventLoop; @@ -39,12 +40,15 @@ import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.Topics; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.ConnectionPool; import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl; import org.awaitility.Awaitility; import org.awaitility.reflect.WhiteboxImpl; @@ -94,9 +98,19 @@ public void testRetryStartProducerStoppedByTopicRemove() throws Exception { when(producerBuilder.create()).thenThrow(new RuntimeException("mocked ex")); when(producerBuilder.createAsync()) .thenReturn(CompletableFuture.failedFuture(new RuntimeException("mocked ex"))); + + @Cleanup + PulsarAdmin admin = mock(PulsarAdmin.class); + Topics adminTopics = mock(Topics.class); + doReturn(adminTopics).when(admin).topics(); + doReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata(0))).when(adminTopics) + .getPartitionedTopicMetadataAsync(anyString()); + doReturn(CompletableFuture.completedFuture(null)).when(adminTopics) + .createNonPartitionedTopicAsync(anyString()); + // Make race condition: "retry start producer" and "close replicator". final ReplicatorInTest replicator = new ReplicatorInTest(localCluster, localTopic, remoteCluster, topicName, - replicatorPrefix, broker, remoteClient); + replicatorPrefix, broker, remoteClient, admin); replicator.startProducer(); replicator.terminate(); @@ -122,9 +136,10 @@ private static class ReplicatorInTest extends AbstractReplicator { public ReplicatorInTest(String localCluster, Topic localTopic, String remoteCluster, String remoteTopicName, String replicatorPrefix, BrokerService brokerService, - PulsarClientImpl replicationClient) throws PulsarServerException { + PulsarClientImpl replicationClient, PulsarAdmin replicationAdmin) + throws PulsarServerException { super(localCluster, localTopic, remoteCluster, remoteTopicName, replicatorPrefix, brokerService, - replicationClient); + replicationClient, replicationAdmin); } @Override 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 8a80f57f5b29f..8971e42f1d11a 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.service; import static org.apache.pulsar.broker.service.persistent.BrokerServicePersistInternalMethodInvoker.ensureNoBacklogByInflightTask; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -63,6 +64,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import lombok.AllArgsConstructor; +import lombok.Cleanup; import lombok.Data; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; @@ -90,6 +92,7 @@ import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.impl.ClientBuilderImpl; import org.apache.pulsar.client.impl.ClientCnx; @@ -108,6 +111,7 @@ import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TopicStats; +import org.apache.pulsar.common.policies.data.TopicType; import org.apache.pulsar.common.policies.data.impl.AutoTopicCreationOverrideImpl; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; @@ -1889,4 +1893,193 @@ public void testReplicatorsInflightTaskListIsEmptyAfterReplicationFinished() thr // Verify: all inflight tasks are done. ensureNoBacklogByInflightTask(getReplicator(topicName)); } + + @DataProvider + public Object[] isPartitioned() { + return new Object[]{ + true, + false + }; + } + + @Test(dataProvider = "isPartitioned") + public void testReplicatorCreateTopic(boolean isPartitioned) throws Exception { + String ns = defaultTenant + "/" + UUID.randomUUID().toString().replace("-", ""); + admin1.namespaces().createNamespace(ns); + if (!usingGlobalZK){ + admin2.namespaces().createNamespace(ns); + } + + int numPartitions = 4; + List partitions = new ArrayList<>(); + final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns + "/tp_"); + if (isPartitioned) { + admin1.topics().createPartitionedTopic(tp, numPartitions); + for (int i = 0; i < numPartitions; i++) { + partitions.add(TopicName.getTopicPartitionNameString(tp, i)); + } + } else { + admin1.topics().createNonPartitionedTopic(tp); + } + + admin1.namespaces().setNamespaceReplicationClusters(ns, new HashSet<>(Arrays.asList(cluster1, cluster2))); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker1.getTopic( + isPartitioned ? TopicName.get(tp).getPartition(0).toString() : tp, false).join() + .get(); + assertFalse(persistentTopic.getReplicators().isEmpty()); + }); + + @Cleanup + Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); + p1.send("msg-1"); + + Awaitility.await().untilAsserted(() -> { + List partitionedTopicList = admin2.topics().getPartitionedTopicList(ns); + if (isPartitioned) { + assertThat(partitionedTopicList).contains(tp); + assertThat(admin2.topics().getList(ns)).containsAll(partitions); + } else { + assertThat(partitionedTopicList).doesNotContain(tp); + assertThat(admin2.topics().getList(ns)).contains(tp); + } + }); + } + + @Test + public void testReplicatorCreateTopicWhenTopicExistsWithDifferentTypeAcrossClusters() throws Exception { + if (usingGlobalZK) { + // This test case is not applicable when using global ZK, because the namespace policies + // are shared among clusters. + return; + } + + String ns = defaultTenant + "/" + UUID.randomUUID().toString().replace("-", ""); + admin1.namespaces().createNamespace(ns); + admin2.namespaces().createNamespace(ns); + + final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns + "/tp_"); + admin1.topics().createPartitionedTopic(tp, 4); + admin2.topics().createNonPartitionedTopic(tp); + + admin1.namespaces().setNamespaceReplicationClusters(ns, new HashSet<>(Arrays.asList(cluster1, cluster2))); + admin2.namespaces().setNamespaceReplicationClusters(ns, new HashSet<>(Arrays.asList(cluster1, cluster2))); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join() + .get(); + assertFalse(persistentTopic.getReplicators().isEmpty()); + }); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = (PersistentTopic) broker2.getTopic(tp, false).join().get(); + assertFalse(persistentTopic.getReplicators().isEmpty()); + }); + + @Cleanup + Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); + p1.send("msg-p1-1"); + @Cleanup + Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); + p2.send("msg-p2-1"); + + // The topic exists, but its type differs between the local and remote clusters. The replicator should not + // recreate the topic. + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join() + .get(); + persistentTopic.getReplicators().forEach((key, value) -> { + assertFalse(value.isConnected()); + }); + }); + assertThat(admin2.topics().getPartitionedTopicList(ns)).doesNotContain(tp); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = (PersistentTopic) broker2.getTopic(tp, false).join().get(); + persistentTopic.getReplicators().forEach((key, value) -> { + assertFalse(value.isConnected()); + }); + }); + assertThat(admin1.topics().getList(ns)).doesNotContain(tp); + } + + @Test + public void testReplicatorWhenPartitionCountsDiffer() throws Exception { + if (usingGlobalZK) { + // This test case is not applicable when using global ZK, because the namespace policies + // are shared among clusters. + return; + } + + String ns = defaultTenant + "/" + UUID.randomUUID().toString().replace("-", ""); + + admin1.namespaces().createNamespace(ns); + admin1.namespaces().setAutoTopicCreation(ns, AutoTopicCreationOverride.builder() + .allowAutoTopicCreation(true) + .topicType(TopicType.PARTITIONED.toString()) + .defaultNumPartitions(12) + .build()); + + admin2.namespaces().createNamespace(ns); + admin2.namespaces().setAutoTopicCreation(ns, AutoTopicCreationOverride.builder() + .allowAutoTopicCreation(true) + .topicType(TopicType.NON_PARTITIONED.toString()) + .build()); + + final String tp = BrokerTestUtil.newUniqueName("persistent://" + ns + "/tp_"); + admin1.topics().createPartitionedTopic(tp, 4); + admin2.topics().createPartitionedTopic(tp, 8); + + admin1.namespaces().setNamespaceReplicationClusters(ns, new HashSet<>(Arrays.asList(cluster1, cluster2))); + admin2.namespaces().setNamespaceReplicationClusters(ns, new HashSet<>(Arrays.asList(cluster1, cluster2))); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join() + .get(); + assertFalse(persistentTopic.getReplicators().isEmpty()); + }); + + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker2.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join() + .get(); + assertFalse(persistentTopic.getReplicators().isEmpty()); + }); + + // Trigger the replicator. + @Cleanup + Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); + p1.send("msg-p1-1"); + @Cleanup + Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); + p2.send("msg-p2-1"); + + // Topic partition counts differ between the local and remote clusters. + // The replicator should not replicate the messages. + Awaitility.await().untilAsserted(() -> { + PersistentTopic persistentTopic = + (PersistentTopic) broker1.getTopic(TopicName.get(tp).getPartition(0).toString(), false).join() + .get(); + persistentTopic.getReplicators().forEach((key, value) -> { + assertFalse(value.isConnected()); + }); + }); + + @Cleanup + Consumer c2 = client2.newConsumer(Schema.STRING).topic(tp).subscriptionName("test-sub") + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe(); + + while (true) { + Message receive = c2.receive(3, TimeUnit.SECONDS); + if (receive == null) { + break; + } + assertEquals(receive.getValue(), "msg-p2-1"); + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index 3b64b2ecc2cd0..ecf3a127576bf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -111,11 +111,14 @@ import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.service.persistent.PulsarCompactorSubscription; import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.Topics; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.ConnectionPool; +import org.apache.pulsar.client.impl.LookupService; import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; @@ -130,6 +133,8 @@ import org.apache.pulsar.common.api.proto.ProducerAccessMode; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; +import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.stats.SubscriptionStatsImpl; @@ -233,6 +238,13 @@ public void setup() throws Exception { doReturn(CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists())).when(nsSvc) .checkTopicExistsAsync(any()); + PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class); + LookupService lookupService = mock(LookupService.class); + doReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata(0))).when(lookupService) + .getPartitionedTopicMetadata(any(), anyBoolean(), anyBoolean()); + doReturn(lookupService).when(pulsarClient).getLookup(); + doReturn(pulsarClient).when(pulsarTestContext.getPulsarService()).getClient(); + setupMLAsyncCallbackMocks(); } @@ -1653,6 +1665,16 @@ public void testFailoverSubscription() throws Exception { assertNull(topic2.getSubscription(successSubName)); } + private PulsarAdmin mockReplicationAdmin() { + PulsarAdmin admin = mock(PulsarAdmin.class); + Topics topics = mock(Topics.class); + doReturn(topics).when(admin).topics(); + doReturn(CompletableFuture.completedFuture(new PartitionedTopicMetadata(0))).when(topics) + .getPartitionedTopicMetadataAsync(anyString()); + doReturn(CompletableFuture.completedFuture(null)).when(topics).createNonPartitionedTopicAsync(anyString()); + return admin; + } + /** * NonPersistentReplicator.removeReplicator doesn't remove replicator in atomic way and does in multiple step: * 1. disconnect replicator producer @@ -1699,11 +1721,16 @@ public CompletableFuture createAsync() { return producerBuilder; }); brokerService.getReplicationClients().put(remoteCluster, pulsarClientMock); + + @Cleanup + PulsarAdmin admin = mockReplicationAdmin(); + brokerService.getClusterAdmins().put(remoteCluster, admin); + Optional clusterData = brokerService.pulsar().getPulsarResources().getClusterResources() + .getCluster(remoteCluster); PersistentReplicator replicator = spy( new GeoPersistentReplicator(topic, cursor, localCluster, remoteCluster, brokerService, - (PulsarClientImpl) brokerService.getReplicationClient(remoteCluster, - brokerService.pulsar().getPulsarResources().getClusterResources() - .getCluster(remoteCluster)))); + (PulsarClientImpl) brokerService.getReplicationClient(remoteCluster, clusterData), + brokerService.getClusterPulsarAdmin(remoteCluster, clusterData))); replicatorMap.put(remoteReplicatorName, replicator); // step-1 remove replicator : it will disconnect the producer but it will wait for callback to be completed @@ -1751,10 +1778,15 @@ public void testClosingReplicationProducerTwice() throws Exception { ManagedCursor cursor = mock(ManagedCursorImpl.class); doReturn(remoteCluster).when(cursor).getName(); brokerService.getReplicationClients().put(remoteCluster, client); + + @Cleanup + PulsarAdmin admin = mockReplicationAdmin(); + brokerService.getClusterAdmins().put(remoteCluster, admin); + Optional clusterData = brokerService.pulsar().getPulsarResources().getClusterResources() + .getCluster(remoteCluster); PersistentReplicator replicator = new GeoPersistentReplicator(topic, cursor, localCluster, remoteCluster, - brokerService, (PulsarClientImpl) brokerService.getReplicationClient(remoteCluster, - brokerService.pulsar().getPulsarResources().getClusterResources() - .getCluster(remoteCluster))); + brokerService, (PulsarClientImpl) brokerService.getReplicationClient(remoteCluster, clusterData), + brokerService.getClusterPulsarAdmin(remoteCluster, clusterData)); // PersistentReplicator constructor calls startProducer() verify(clientImpl) From e64cd5e150fd8b22cce27f80f948a0dc42753cf6 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 25 Nov 2025 12:00:06 +0800 Subject: [PATCH 2/6] [fix][broker] Update NonPersistentReplicator to include PulsarAdmin in constructor --- .../broker/service/nonpersistent/NonPersistentReplicator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java index 076c0aa29353d..38320e5be70c5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java @@ -55,7 +55,7 @@ public NonPersistentReplicator(NonPersistentTopic topic, String localCluster, St BrokerService brokerService, PulsarClientImpl replicationClient, PulsarAdmin replicationAdmin) throws PulsarServerException { super(localCluster, topic, remoteCluster, topic.getName(), topic.getReplicatorPrefix(), brokerService, - replicationClient); + replicationClient, replicationAdmin); // NonPersistentReplicator does not support limitation so far, so reset pending queue size to the default value. producerBuilder.maxPendingMessages(1000); producerBuilder.blockIfQueueFull(false); From d5cdce6b4ba905962db7c0b3ab42240ccb9f0c9f Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Fri, 28 Nov 2025 16:56:54 +0800 Subject: [PATCH 3/6] address comments --- .../broker/service/AbstractReplicator.java | 2 + .../persistent/GeoPersistentReplicator.java | 234 ++++++++---------- .../broker/service/OneWayReplicatorTest.java | 102 ++++++++ 3 files changed, 208 insertions(+), 130 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java index 6975ba86e8d90..c7a36ad1b21d7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java @@ -59,6 +59,7 @@ public abstract class AbstractReplicator implements Replicator { protected final PulsarClientImpl replicationClient; protected final PulsarClientImpl client; protected final PulsarAdmin replicationAdmin; + protected final PulsarAdmin admin; protected String replicatorId; @Getter protected final Topic localTopic; @@ -122,6 +123,7 @@ public AbstractReplicator(String localCluster, Topic localTopic, String remoteCl this.replicationClient = replicationClient; this.replicationAdmin = replicationAdmin; this.client = (PulsarClientImpl) brokerService.pulsar().getClient(); + this.admin = brokerService.pulsar().getAdminClient(); this.producer = null; this.producerQueueSize = brokerService.pulsar().getConfiguration().getReplicationProducerQueueSize(); this.replicatorId = String.format("%s | %s", diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index 9ee84f0c7ce8b..3e8380d6e85c7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -22,21 +22,17 @@ import io.netty.buffer.ByteBuf; import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.client.admin.PulsarAdmin; -import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException; -import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException; -import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.protocol.Markers; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.util.FutureUtil; @@ -62,137 +58,115 @@ protected String getProducerName() { @Override protected CompletableFuture prepareCreateProducer() { - if (brokerService.getPulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) { - TopicName completeTopicName = TopicName.get(localTopicName); - TopicName baseTopicName; - if (completeTopicName.isPartitioned()) { - baseTopicName = TopicName.get(completeTopicName.getPartitionedTopicName()); - } else { - baseTopicName = completeTopicName; - } - // Set useFallbackForNonPIP344Brokers to true when mix of PIP-344 and non-PIP-344 brokers are used, it - // can still work. - return client.getLookup().getPartitionedTopicMetadata(baseTopicName, false, true) - .thenCompose((localMetadata) -> replicationAdmin.topics() - // https://github.com/apache/pulsar/pull/4963 - // Use the admin API instead of the client to fetch partitioned metadata - // to prevent automatic topic creation on the remote cluster. - // PIP-344 introduced an option to disable auto-creation when fetching partitioned - // topic metadata via the client, but this requires Pulsar 3.0.x. - // This change is a workaround to support Pulsar 2.4.2. - .getPartitionedTopicMetadataAsync(baseTopicName.toString()) - .exceptionally(ex -> { - Throwable throwable = FutureUtil.unwrapCompletionException(ex); - if (throwable instanceof NotFoundException) { - // Topic does not exist on the remote cluster. - return new PartitionedTopicMetadata(0); - } - throw new CompletionException("Failed to get partitioned topic metadata", throwable); - }).thenCompose(remoteMetadata -> { - if (log.isDebugEnabled()) { - log.debug("[{}] Local metadata partitions: {} Remote metadata partitions: {}", - replicatorId, localMetadata.partitions, remoteMetadata.partitions); - } + return createRemoteTopicIfDoesNotExist(TopicName.get(localTopicName).getPartitionedTopicName()); + } - // Non-partitioned topic - if (localMetadata.partitions == 0) { - if (localMetadata.partitions == remoteMetadata.partitions) { - return replicationAdmin.topics().createNonPartitionedTopicAsync(localTopicName) - .exceptionally(ex -> { - Throwable throwable = FutureUtil.unwrapCompletionException(ex); - if (throwable instanceof ConflictException) { - // Topic already exists on the remote cluster. - return null; - } else { - throw new CompletionException( - "Failed to create non-partitioned topic", throwable); - } - }); - } else { - return FutureUtil.failedFuture(new PulsarClientException.NotAllowedException( - "Topic type is not matched between local and remote cluster: local " - + "partitions: " + localMetadata.partitions - + ", remote partitions: " + remoteMetadata.partitions)); - } + private CompletableFuture createRemoteTopicIfDoesNotExist(String partitionedTopic) { + CompletableFuture res = new CompletableFuture<>(); + admin.topics().getPartitionedTopicMetadataAsync(partitionedTopic).whenComplete((local, t1) -> { + if (t1 != null) { + Throwable actEx = FutureUtil.unwrapCompletionException(t1); + // Local topic is a non-partitioned topic, but end with "-partition-{num}". + if (actEx instanceof PulsarAdminException.NotFoundException) { + replicationAdmin.topics().getPartitionedTopicMetadataAsync(partitionedTopic) + .whenComplete((remote, t2) -> { + if (t2 != null) { + Throwable actEx2 = FutureUtil.unwrapCompletionException(t2); + if (actEx2 instanceof PulsarAdminException.NotFoundException) { + // Both clusters have a non-partitioned topic, but the topic name is end with + // "-partition-{num}". + // Check partition metadata with the special name. + FutureUtil.completeAfter(res, createRemoteTopicIfDoesNotExist(localTopicName)); } else { - if (remoteMetadata.partitions == 0) { - if (log.isDebugEnabled()) { - log.debug("[{}] Creating partitioned topic {} with {} partitions", - replicatorId, baseTopicName, localMetadata.partitions); - } - // We maybe need to create a partitioned topic on remote cluster. - return replicationAdmin.topics() - .createPartitionedTopicAsync(baseTopicName.toString(), - localMetadata.partitions) - .exceptionally(ex -> { - Throwable throwable = FutureUtil.unwrapCompletionException(ex); - if (throwable instanceof ConflictException) { - // Topic already exists on the remote cluster. - // This can happen if the topic was created, or the topic is - // non-partitioned. - return null; - } else { - throw new CompletionException( - "Failed to create partitioned topic", throwable); - } - }) - .thenCompose((__) -> replicationAdmin.topics() - .getPartitionedTopicMetadataAsync(baseTopicName.toString())) - .thenCompose(metadata -> { - // Double check if the partitioned topic is created - // successfully. - // When partitions is equals to 0, it means this topic is - // non-partitioned, we should throw an exception. - if (completeTopicName.getPartitionIndex() >= metadata.partitions) { - return FutureUtil.failedFuture( - new PulsarClientException.NotAllowedException( - "Topic type is not matched between " - + "local and " - + "remote cluster: local " - + "partitions: " - + localMetadata.partitions - + ", remote partitions: " - + remoteMetadata.partitions)); - } - return CompletableFuture.completedFuture(null); - }); - } else { - if (localMetadata.partitions != remoteMetadata.partitions) { - return FutureUtil.failedFuture( - new PulsarClientException.NotAllowedException( - "The number of topic partitions is inconsistent between " - + "local and" - + " remote " - + "clusters: local partitions: " - + localMetadata.partitions - + ", remote partitions: " - + remoteMetadata.partitions)); - } - } + // Failed to get remote partitions. + String errorMsg = String.format("[%s] Can not start replicator because of failed to" + + " get topic partitions of remote cluster. The topic on the local cluster is" + + " a non-partitioned topic, but end with -partition-x", + replicatorId); + log.error(errorMsg, actEx); + res.completeExceptionally(new PulsarServerException(errorMsg)); + return; } - return CompletableFuture.completedFuture(null); - })); - } else { - CompletableFuture topicCheckFuture = new CompletableFuture<>(); - replicationClient.getPartitionedTopicMetadata(localTopic.getName(), false, false) - .whenComplete((metadata, ex) -> { - if (ex == null) { - if (metadata.partitions == 0) { - topicCheckFuture.complete(null); - } else { - String errorMsg = String.format("%s Can not create the replicator due to the partitions in the" - + " remote cluster is not 0, but is %s", - replicatorId, metadata.partitions); - log.error(errorMsg); - topicCheckFuture.completeExceptionally( - new PulsarClientException.NotAllowedException(errorMsg)); + return; + } + // Local topic is a non-partitioned topic, but end with "-partition-{num}". + // Remote side: it has a partitioned topic. + String errorMsg = String.format("[%s] Can not start replicator because the" + + " partitions between local and remote cluster are different." + + " The topic on the local cluster is a non-partitioned topic, but end with" + + " -partition-x, and remote side it has %s partitions", + replicatorId, remote.partitions); + log.error(errorMsg); + res.completeExceptionally(new PulsarServerException(errorMsg)); + }); + return; + } + // Failed to get local partitions. + log.error("[{}] Failed to start replicator because of failed to get partitions of local cluster. The" + + " topic on the local cluster has {} partitions", + replicatorId, local.partitions, actEx); + res.completeExceptionally(actEx); + return; + } + replicationAdmin.topics().getPartitionedTopicMetadataAsync(partitionedTopic).whenComplete((remote, t2) -> { + if (t2 != null) { + Throwable actEx = FutureUtil.unwrapCompletionException(t2); + // Create the topic on the remote side. + if (actEx instanceof PulsarAdminException.NotFoundException) { + // Not allowed replicator to create topics on the remote side. + if (!brokerService.getPulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) { + String errorMsg = String.format("[%s] Can not start replicator because there is no topic on" + + " the remote cluster. Please create a %s on the remote cluster", + replicatorId, local.partitions == 0 ? "non-partitioned topic" + : "partitioned topic with " + local.partitions + " partitions"); + log.error(errorMsg); + res.completeExceptionally(new PulsarServerException(errorMsg)); + return; + } + // Create non-partitioned topics. + // Print errors if failed to create topocs. + res.whenComplete((__, t) -> { + if (t == null) { + return; + } + // Failed to get remote partitions. + log.error("[{}] Failed to start replicator because of failed to" + + " create topic on the remote cluster. The topic on the local cluster has" + + " {} partitions", replicatorId, local.partitions, t); + }); + if (local.partitions == 0) { + FutureUtil.completeAfter(res, replicationAdmin.topics() + .createNonPartitionedTopicAsync(partitionedTopic)); + return; + } + // Create partitioned topics. + FutureUtil.completeAfter(res, replicationAdmin.topics() + .createPartitionedTopicAsync(partitionedTopic, local.partitions)); + return; } - } else { - topicCheckFuture.completeExceptionally(FutureUtil.unwrapCompletionException(ex)); + // Failed to get remote partitions. + String errorMsg = String.format("[%s] Can not start replicator because of failed to get" + + " topic partitions of remote cluster. The topic on the local cluster has" + + " %s partitions", + replicatorId, local.partitions); + log.error(errorMsg, actEx); + res.completeExceptionally(new PulsarServerException(errorMsg)); + return; + } + // Compacted partitions. + if (local.partitions == remote.partitions) { + res.complete(null); + return; } + // Incompatible partitions between clusters. + String errorMsg = String.format("[%s] Can not start replicator because the partitions between" + + " local and remote cluster are different. local: %s, remote: %s", replicatorId, + local.partitions, remote.partitions); + log.error(errorMsg); + res.completeExceptionally(new PulsarServerException(errorMsg)); }); - return topicCheckFuture; - } + }); + return res; } @Override 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 8971e42f1d11a..169c91ead8904 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 @@ -2082,4 +2082,106 @@ public void testReplicatorWhenPartitionCountsDiffer() throws Exception { assertEquals(receive.getValue(), "msg-p2-1"); } } + + // https://github.com/apache/pulsar/pull/24118 has denied a client to access a non-partitioned topic that end with + // "-partition-{num}", but it is not cherry-picked to previous branches. + // To guarantee compatibility with previous releases, the following tests should work for the releases that is + // earlier than "4.1.0". + // When cherry-picking the PR into previous branches, the following tests should be enabled. + // I will push a seperate PR to remove these tests. +// @Test +// public void testReplicatorWithSpecialNonPartitionedTopic() throws Exception { +// if (usingGlobalZK) { +// // This test case is not applicable when using global ZK, because the namespace policies +// // are shared among clusters. +// return; +// } +// +// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-4"; +// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); +// pulsar1.getDefaultManagedLedgerFactory().open(mlName); +// pulsar2.getDefaultManagedLedgerFactory().open(mlName); +// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); +// Consumer c2 = client2.newConsumer(Schema.STRING).topic(tp).subscriptionName("test-sub") +// .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe(); +// +// waitReplicatorStarted(tp); +// p1.send("abc"); +// Message msg = c2.receive(2, TimeUnit.SECONDS); +// assertNotNull(msg); +// assertEquals(msg.getValue(), "abc"); +// +// // cleanup +// p1.close(); +// c2.close(); +// } +// +// @Test +// public void testFailureReplicatorWithSpecialNonPartitionedTopic() throws Exception { +// if (usingGlobalZK) { +// // This test case is not applicable when using global ZK, because the namespace policies +// // are shared among clusters. +// return; +// } +// +// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-0"; +// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); +// pulsar2.getDefaultManagedLedgerFactory().open(mlName); +// Thread.sleep(3000); +// admin1.topics().createPartitionedTopic(TopicName.get(tp).getPartitionedTopicName(), 1); +// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); +// Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); +// p1.send("123"); +// p2.send("321"); +// Thread.sleep(3000); +// +// Awaitility.await().untilAsserted(() -> { +// PersistentTopic persistentTopic = (PersistentTopic) broker1 +// .getTopic(TopicName.get(tp).getPartition(0).toString(), false).join().get(); +// assertTrue(persistentTopic.getReplicators().isEmpty() +// || !persistentTopic.getReplicators().get(cluster2).isConnected()); +// }); +// +// // cleanup +// p1.close(); +// p2.close(); +// cleanupTopics(() -> { +// admin1.topics().deletePartitionedTopic(TopicName.get(tp).getPartitionedTopicName()); +// admin2.topics().delete(tp); +// }); +// } +// +// @Test +// public void testFailureReplicatorWithSpecialNonPartitionedTopic2() throws Exception { +// if (usingGlobalZK) { +// // This test case is not applicable when using global ZK, because the namespace policies +// // are shared among clusters. +// return; +// } +// +// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-0"; +// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); +// pulsar1.getDefaultManagedLedgerFactory().open(mlName); +// Thread.sleep(3000); +// admin2.topics().createPartitionedTopic(TopicName.get(tp).getPartitionedTopicName(), 1); +// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); +// Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); +// p1.send("123"); +// p2.send("321"); +// Thread.sleep(3000); +// +// Awaitility.await().untilAsserted(() -> { +// PersistentTopic persistentTopic = (PersistentTopic) broker1.getTopic(tp, false).join().get(); +// assertTrue(persistentTopic.getReplicators().isEmpty() +// || !persistentTopic.getReplicators().get(cluster2).isConnected()); +// }); +// +// // cleanup +// p1.close(); +// p2.close(); +// cleanupTopics(() -> { +// admin1.topics().delete(tp); +// admin2.topics().deletePartitionedTopic(TopicName.get(tp).getPartitionedTopicName()); +// }); +// } } From 00a52362b346277868473ca7a7183cbc5f0bfd60 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Thu, 25 Dec 2025 12:16:11 +0800 Subject: [PATCH 4/6] [fix][test] Remove obsolete tests --- .../broker/service/OneWayReplicatorTest.java | 102 ------------------ 1 file changed, 102 deletions(-) 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 169c91ead8904..8971e42f1d11a 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 @@ -2082,106 +2082,4 @@ public void testReplicatorWhenPartitionCountsDiffer() throws Exception { assertEquals(receive.getValue(), "msg-p2-1"); } } - - // https://github.com/apache/pulsar/pull/24118 has denied a client to access a non-partitioned topic that end with - // "-partition-{num}", but it is not cherry-picked to previous branches. - // To guarantee compatibility with previous releases, the following tests should work for the releases that is - // earlier than "4.1.0". - // When cherry-picking the PR into previous branches, the following tests should be enabled. - // I will push a seperate PR to remove these tests. -// @Test -// public void testReplicatorWithSpecialNonPartitionedTopic() throws Exception { -// if (usingGlobalZK) { -// // This test case is not applicable when using global ZK, because the namespace policies -// // are shared among clusters. -// return; -// } -// -// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-4"; -// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); -// pulsar1.getDefaultManagedLedgerFactory().open(mlName); -// pulsar2.getDefaultManagedLedgerFactory().open(mlName); -// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); -// Consumer c2 = client2.newConsumer(Schema.STRING).topic(tp).subscriptionName("test-sub") -// .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe(); -// -// waitReplicatorStarted(tp); -// p1.send("abc"); -// Message msg = c2.receive(2, TimeUnit.SECONDS); -// assertNotNull(msg); -// assertEquals(msg.getValue(), "abc"); -// -// // cleanup -// p1.close(); -// c2.close(); -// } -// -// @Test -// public void testFailureReplicatorWithSpecialNonPartitionedTopic() throws Exception { -// if (usingGlobalZK) { -// // This test case is not applicable when using global ZK, because the namespace policies -// // are shared among clusters. -// return; -// } -// -// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-0"; -// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); -// pulsar2.getDefaultManagedLedgerFactory().open(mlName); -// Thread.sleep(3000); -// admin1.topics().createPartitionedTopic(TopicName.get(tp).getPartitionedTopicName(), 1); -// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); -// Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); -// p1.send("123"); -// p2.send("321"); -// Thread.sleep(3000); -// -// Awaitility.await().untilAsserted(() -> { -// PersistentTopic persistentTopic = (PersistentTopic) broker1 -// .getTopic(TopicName.get(tp).getPartition(0).toString(), false).join().get(); -// assertTrue(persistentTopic.getReplicators().isEmpty() -// || !persistentTopic.getReplicators().get(cluster2).isConnected()); -// }); -// -// // cleanup -// p1.close(); -// p2.close(); -// cleanupTopics(() -> { -// admin1.topics().deletePartitionedTopic(TopicName.get(tp).getPartitionedTopicName()); -// admin2.topics().delete(tp); -// }); -// } -// -// @Test -// public void testFailureReplicatorWithSpecialNonPartitionedTopic2() throws Exception { -// if (usingGlobalZK) { -// // This test case is not applicable when using global ZK, because the namespace policies -// // are shared among clusters. -// return; -// } -// -// String tp = BrokerTestUtil.newUniqueName("persistent://" + replicatedNamespace + "/tp") + "-partition-0"; -// String mlName = TopicName.get(tp).getPersistenceNamingEncoding(); -// pulsar1.getDefaultManagedLedgerFactory().open(mlName); -// Thread.sleep(3000); -// admin2.topics().createPartitionedTopic(TopicName.get(tp).getPartitionedTopicName(), 1); -// Producer p1 = client1.newProducer(Schema.STRING).topic(tp).create(); -// Producer p2 = client2.newProducer(Schema.STRING).topic(tp).create(); -// p1.send("123"); -// p2.send("321"); -// Thread.sleep(3000); -// -// Awaitility.await().untilAsserted(() -> { -// PersistentTopic persistentTopic = (PersistentTopic) broker1.getTopic(tp, false).join().get(); -// assertTrue(persistentTopic.getReplicators().isEmpty() -// || !persistentTopic.getReplicators().get(cluster2).isConnected()); -// }); -// -// // cleanup -// p1.close(); -// p2.close(); -// cleanupTopics(() -> { -// admin1.topics().delete(tp); -// admin2.topics().deletePartitionedTopic(TopicName.get(tp).getPartitionedTopicName()); -// }); -// } } From 3867f5196c33cb6758007c51fd2fb565d9425143 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Thu, 25 Dec 2025 16:51:23 +0800 Subject: [PATCH 5/6] [improve][broker] Refactor topic creation logic --- .../persistent/GeoPersistentReplicator.java | 174 ++++++++---------- 1 file changed, 72 insertions(+), 102 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index 3e8380d6e85c7..922a0e42c3ddb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -61,112 +61,82 @@ protected CompletableFuture prepareCreateProducer() { return createRemoteTopicIfDoesNotExist(TopicName.get(localTopicName).getPartitionedTopicName()); } - private CompletableFuture createRemoteTopicIfDoesNotExist(String partitionedTopic) { - CompletableFuture res = new CompletableFuture<>(); - admin.topics().getPartitionedTopicMetadataAsync(partitionedTopic).whenComplete((local, t1) -> { - if (t1 != null) { - Throwable actEx = FutureUtil.unwrapCompletionException(t1); - // Local topic is a non-partitioned topic, but end with "-partition-{num}". - if (actEx instanceof PulsarAdminException.NotFoundException) { - replicationAdmin.topics().getPartitionedTopicMetadataAsync(partitionedTopic) - .whenComplete((remote, t2) -> { - if (t2 != null) { - Throwable actEx2 = FutureUtil.unwrapCompletionException(t2); - if (actEx2 instanceof PulsarAdminException.NotFoundException) { - // Both clusters have a non-partitioned topic, but the topic name is end with - // "-partition-{num}". - // Check partition metadata with the special name. - FutureUtil.completeAfter(res, createRemoteTopicIfDoesNotExist(localTopicName)); - } else { - // Failed to get remote partitions. - String errorMsg = String.format("[%s] Can not start replicator because of failed to" - + " get topic partitions of remote cluster. The topic on the local cluster is" - + " a non-partitioned topic, but end with -partition-x", - replicatorId); - log.error(errorMsg, actEx); - res.completeExceptionally(new PulsarServerException(errorMsg)); - return; - } - return; - } - // Local topic is a non-partitioned topic, but end with "-partition-{num}". - // Remote side: it has a partitioned topic. - String errorMsg = String.format("[%s] Can not start replicator because the" - + " partitions between local and remote cluster are different." - + " The topic on the local cluster is a non-partitioned topic, but end with" - + " -partition-x, and remote side it has %s partitions", - replicatorId, remote.partitions); - log.error(errorMsg); - res.completeExceptionally(new PulsarServerException(errorMsg)); - }); - return; - } - // Failed to get local partitions. - log.error("[{}] Failed to start replicator because of failed to get partitions of local cluster. The" - + " topic on the local cluster has {} partitions", - replicatorId, local.partitions, actEx); - res.completeExceptionally(actEx); - return; - } - replicationAdmin.topics().getPartitionedTopicMetadataAsync(partitionedTopic).whenComplete((remote, t2) -> { - if (t2 != null) { - Throwable actEx = FutureUtil.unwrapCompletionException(t2); - // Create the topic on the remote side. + private CompletableFuture getLocalPartitionMetadata(String topic) { + return admin.topics().getPartitionedTopicMetadataAsync(topic) + .thenApply(metadata -> metadata.partitions) + .exceptionallyCompose(t -> { + Throwable actEx = FutureUtil.unwrapCompletionException(t); if (actEx instanceof PulsarAdminException.NotFoundException) { - // Not allowed replicator to create topics on the remote side. - if (!brokerService.getPulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) { - String errorMsg = String.format("[%s] Can not start replicator because there is no topic on" - + " the remote cluster. Please create a %s on the remote cluster", - replicatorId, local.partitions == 0 ? "non-partitioned topic" - : "partitioned topic with " + local.partitions + " partitions"); - log.error(errorMsg); - res.completeExceptionally(new PulsarServerException(errorMsg)); - return; - } - // Create non-partitioned topics. - // Print errors if failed to create topocs. - res.whenComplete((__, t) -> { - if (t == null) { - return; - } - // Failed to get remote partitions. - log.error("[{}] Failed to start replicator because of failed to" - + " create topic on the remote cluster. The topic on the local cluster has" - + " {} partitions", replicatorId, local.partitions, t); - }); - if (local.partitions == 0) { - FutureUtil.completeAfter(res, replicationAdmin.topics() - .createNonPartitionedTopicAsync(partitionedTopic)); - return; - } - // Create partitioned topics. - FutureUtil.completeAfter(res, replicationAdmin.topics() - .createPartitionedTopicAsync(partitionedTopic, local.partitions)); - return; + // Legacy edge case: Local topic is non-partitioned but name ends with "-partition-{num}". + // This should never happen in practice because PIP-414 disables this naming pattern. + return createRemoteTopicIfDoesNotExist(localTopicName) + .thenApply(__ -> -1); // Special marker } - // Failed to get remote partitions. - String errorMsg = String.format("[%s] Can not start replicator because of failed to get" - + " topic partitions of remote cluster. The topic on the local cluster has" - + " %s partitions", - replicatorId, local.partitions); - log.error(errorMsg, actEx); - res.completeExceptionally(new PulsarServerException(errorMsg)); - return; - } - // Compacted partitions. - if (local.partitions == remote.partitions) { - res.complete(null); - return; - } - // Incompatible partitions between clusters. - String errorMsg = String.format("[%s] Can not start replicator because the partitions between" - + " local and remote cluster are different. local: %s, remote: %s", replicatorId, - local.partitions, remote.partitions); + return CompletableFuture.failedFuture(actEx); + }); + } + + private CompletableFuture getRemotePartitionMetadata(String topic) { + return replicationAdmin.topics().getPartitionedTopicMetadataAsync(topic) + .thenApply(metadata -> metadata.partitions) + .exceptionallyCompose(t -> { + Throwable actEx = FutureUtil.unwrapCompletionException(t); + if (actEx instanceof PulsarAdminException.NotFoundException) { + return CompletableFuture.completedFuture(-1); // Topic doesn't exist + } + return CompletableFuture.failedFuture(actEx); + }); + } + + private CompletableFuture handlePartitionComparison(String topic, int localPartitions, int remotePartitions) { + // Skip if already handled by recursion + if (localPartitions == -1) { + return CompletableFuture.completedFuture(null); + } + + // Remote topic doesn't exist - create it + if (remotePartitions == -1) { + if (!brokerService.getPulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) { + String errorMsg = String.format("[%s] Can not start replicator because there is no topic on" + + " the remote cluster. Please create a %s on the remote cluster", + replicatorId, localPartitions == 0 ? "non-partitioned topic" + : "partitioned topic with " + localPartitions + " partitions"); log.error(errorMsg); - res.completeExceptionally(new PulsarServerException(errorMsg)); + return CompletableFuture.failedFuture(new PulsarServerException(errorMsg)); + } + + CompletableFuture createFuture = localPartitions == 0 + ? replicationAdmin.topics().createNonPartitionedTopicAsync(topic) + : replicationAdmin.topics().createPartitionedTopicAsync(topic, localPartitions); + + return createFuture.whenComplete((__, t) -> { + if (t != null) { + log.error("[{}] Failed to create topic on remote cluster. Local has {} partitions", + replicatorId, localPartitions, t); + } }); - }); - return res; + } + + // Both exist - verify compatibility + if (localPartitions == remotePartitions) { + return CompletableFuture.completedFuture(null); + } + + // Incompatible partitions + String errorMsg = String.format("[%s] Can not start replicator because the partitions between" + + " local and remote cluster are different. local: %s, remote: %s", + replicatorId, localPartitions, remotePartitions); + log.error(errorMsg); + return CompletableFuture.failedFuture(new PulsarServerException(errorMsg)); + } + + private CompletableFuture createRemoteTopicIfDoesNotExist(String partitionedTopic) { + return getLocalPartitionMetadata(partitionedTopic) + .thenCompose(localPartitions -> + getRemotePartitionMetadata(partitionedTopic).thenCompose(remotePartitions -> + handlePartitionComparison(partitionedTopic, localPartitions, remotePartitions) + ) + ); } @Override From b5953d96867f3c45cc14461cc87d9a2c7bb54009 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Fri, 9 Jan 2026 18:37:18 +0800 Subject: [PATCH 6/6] Fix test --- .../org/apache/pulsar/broker/service/PersistentTopicTest.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index ecf3a127576bf..c83f3750ff03f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -1724,6 +1724,8 @@ public CompletableFuture createAsync() { @Cleanup PulsarAdmin admin = mockReplicationAdmin(); + PulsarService pulsar = brokerService.getPulsar(); + doReturn(admin).when(pulsar).getAdminClient(); brokerService.getClusterAdmins().put(remoteCluster, admin); Optional clusterData = brokerService.pulsar().getPulsarResources().getClusterResources() .getCluster(remoteCluster); @@ -1781,6 +1783,7 @@ public void testClosingReplicationProducerTwice() throws Exception { @Cleanup PulsarAdmin admin = mockReplicationAdmin(); + doReturn(admin).when(pulsar).getAdminClient(); brokerService.getClusterAdmins().put(remoteCluster, admin); Optional clusterData = brokerService.pulsar().getPulsarResources().getClusterResources() .getCluster(remoteCluster);