You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rd...@apache.org on 2023/10/16 18:26:37 UTC
[pulsar] branch master updated: [fix][broker] Fix avoid creating new topic after migration is started (#21368)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e08841114ce [fix][broker] Fix avoid creating new topic after migration is started (#21368)
e08841114ce is described below
commit e08841114cece7e4dc64677505ab66271a2bace5
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Mon Oct 16 11:26:29 2023 -0700
[fix][broker] Fix avoid creating new topic after migration is started (#21368)
Co-authored-by: Rajan Dhabalia <rd...@oath.com>
---
.../pulsar/broker/service/AbstractTopic.java | 5 +++
.../pulsar/broker/service/BrokerService.java | 26 ++++++++++-
.../apache/pulsar/broker/service/ServerCnx.java | 51 +++++++++++++++++++---
.../broker/service/ClusterMigrationTest.java | 43 +++++++++++++++---
4 files changed, 114 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 3cb396d7a4b..a8f25f61a94 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -1352,6 +1352,11 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
return getMigratedClusterUrl(brokerService.getPulsar(), topic);
}
+ public static CompletableFuture<Boolean> isClusterMigrationEnabled(PulsarService pulsar,
+ String topic) {
+ return getMigratedClusterUrlAsync(pulsar, topic).thenApply(url -> url.isPresent());
+ }
+
public static CompletableFuture<Optional<ClusterUrl>> getMigratedClusterUrlAsync(PulsarService pulsar,
String topic) {
return pulsar.getPulsarResources().getClusterResources().getClusterAsync(pulsar.getConfig().getClusterName())
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 d03a94a0563..b85f77cb2f5 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
@@ -111,6 +111,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.NamingException;
import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
+import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
@@ -1521,6 +1522,12 @@ public class BrokerService implements Closeable {
topicFuture.handle((persistentTopic, ex) -> {
// release permit and process pending topic
topicLoadSemaphore.release();
+ // do not recreate topic if topic is already migrated and deleted by broker
+ // so, avoid creating a new topic if migration is already started
+ if (ex != null && (ex.getCause() instanceof TopicMigratedException)) {
+ topicFuture.completeExceptionally(ex.getCause());
+ return null;
+ }
createPendingLoadTopic();
return null;
});
@@ -1632,7 +1639,10 @@ public class BrokerService implements Closeable {
? checkMaxTopicsPerNamespace(topicName, 1)
: CompletableFuture.completedFuture(null);
- maxTopicsCheck.thenCompose(__ -> getManagedLedgerConfig(topicName)).thenAccept(managedLedgerConfig -> {
+ CompletableFuture<Void> isTopicAlreadyMigrated = checkTopicAlreadyMigrated(topicName);
+
+ maxTopicsCheck.thenCompose(__ -> isTopicAlreadyMigrated).thenCompose(__ -> getManagedLedgerConfig(topicName))
+ .thenAccept(managedLedgerConfig -> {
if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) {
// init managedLedger interceptor
Set<BrokerEntryMetadataInterceptor> interceptors = new HashSet<>();
@@ -1760,6 +1770,20 @@ public class BrokerService implements Closeable {
});
}
+ private CompletableFuture<Void> checkTopicAlreadyMigrated(TopicName topicName) {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ AbstractTopic.isClusterMigrationEnabled(pulsar, topicName.toString()).handle((isMigrated, ex) -> {
+ if (isMigrated) {
+ result.completeExceptionally(
+ new BrokerServiceException.TopicMigratedException(topicName + " already migrated"));
+ } else {
+ result.complete(null);
+ }
+ return null;
+ });
+ return result;
+ }
+
public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@Nonnull TopicName topicName) {
requireNonNull(topicName);
NamespaceName namespace = topicName.getNamespaceObject();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 18bf62f1ef3..efbe3fcd7ec 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1304,6 +1304,24 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
remoteAddress, topicName, subscriptionName,
exception.getCause().getMessage());
}
+ } else if (exception.getCause() instanceof BrokerServiceException.TopicMigratedException) {
+ Optional<ClusterUrl> clusterURL = getMigratedClusterUrl(service.getPulsar(),
+ topicName.toString());
+ if (clusterURL.isPresent()) {
+ log.info("[{}] redirect migrated consumer to topic {}: "
+ + "consumerId={}, subName={}, {}", remoteAddress,
+ topicName, consumerId, subscriptionName, exception.getCause().getMessage());
+ boolean msgSent = commandSender.sendTopicMigrated(ResourceType.Consumer, consumerId,
+ clusterURL.get().getBrokerServiceUrl(),
+ clusterURL.get().getBrokerServiceUrlTls());
+ if (!msgSent) {
+ log.info("consumer client doesn't support topic migration handling {}-{}-{}",
+ topicName, remoteAddress, consumerId);
+ }
+ consumers.remove(consumerId, consumerFuture);
+ closeConsumer(consumerId);
+ return null;
+ }
} else if (exception.getCause() instanceof BrokerServiceException) {
log.warn("[{}][{}][{}] Failed to create consumer: consumerId={}, {}",
remoteAddress, topicName, subscriptionName,
@@ -1567,6 +1585,22 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
producers.remove(producerId, producerFuture);
return null;
+ } else if (cause instanceof BrokerServiceException.TopicMigratedException) {
+ Optional<ClusterUrl> clusterURL = getMigratedClusterUrl(service.getPulsar(), topicName.toString());
+ if (clusterURL.isPresent()) {
+ log.info("[{}] redirect migrated producer to topic {}: "
+ + "producerId={}, producerName = {}, {}", remoteAddress,
+ topicName, producerId, producerName, cause.getMessage());
+ boolean msgSent = commandSender.sendTopicMigrated(ResourceType.Producer, producerId,
+ clusterURL.get().getBrokerServiceUrl(), clusterURL.get().getBrokerServiceUrlTls());
+ if (!msgSent) {
+ log.info("client doesn't support topic migration handling {}-{}-{}", topicName,
+ remoteAddress, producerId);
+ }
+ producers.remove(producerId, producerFuture);
+ closeProducer(producerId, -1L);
+ return null;
+ }
}
// Do not print stack traces for expected exceptions
@@ -2986,15 +3020,18 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
public void closeProducer(Producer producer) {
// removes producer-connection from map and send close command to producer
safelyRemoveProducer(producer);
+ closeProducer(producer.getProducerId(), producer.getEpoch());
+
+ }
+
+ public void closeProducer(long producerId, long epoch) {
if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {
- writeAndFlush(Commands.newCloseProducer(producer.getProducerId(), -1L));
+ writeAndFlush(Commands.newCloseProducer(producerId, -1L));
// The client does not necessarily know that the producer is closed, but the connection is still
// active, and there could be messages in flight already. We want to ignore these messages for a time
// because they are expected. Once the interval has passed, the client should have received the
// CloseProducer command and should not send any additional messages until it sends a create Producer
// command.
- final long epoch = producer.getEpoch();
- final long producerId = producer.getProducerId();
recentlyClosedProducers.put(producerId, epoch);
ctx.executor().schedule(() -> {
recentlyClosedProducers.remove(producerId, epoch);
@@ -3009,8 +3046,12 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
public void closeConsumer(Consumer consumer) {
// removes consumer-connection from map and send close command to consumer
safelyRemoveConsumer(consumer);
+ closeConsumer(consumer.consumerId());
+ }
+
+ public void closeConsumer(long consumerId) {
if (getRemoteEndpointProtocolVersion() >= v5.getValue()) {
- writeAndFlush(Commands.newCloseConsumer(consumer.consumerId(), -1L));
+ writeAndFlush(Commands.newCloseConsumer(consumerId, -1L));
} else {
close();
}
@@ -3529,4 +3570,4 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
protected void setAuthRole(String authRole) {
this.authRole = authRole;
}
-}
+}
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
index a2f32170f0f..f2e0cf7fd12 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
@@ -368,8 +368,8 @@ public class ClusterMigrationTest {
String newTopicName = topicName + "-new";
consumerM = client1.newConsumer().topic(newTopicName).subscriptionType(subType)
.subscriptionName("sM").subscribe();
- retryStrategically((t) -> pulsar1.getBrokerService().getTopicReference(newTopicName).isPresent(), 5, 100);
- pulsar1.getBrokerService().getTopicReference(newTopicName).get().checkClusterMigration().get();
+ retryStrategically((t) -> pulsar2.getBrokerService().getTopicReference(newTopicName).isPresent(), 5, 100);
+ pulsar2.getBrokerService().getTopicReference(newTopicName).get().checkClusterMigration().get();
retryStrategically((t) ->
pulsar2.getBrokerService().getTopicReference(newTopicName).isPresent() &&
pulsar2.getBrokerService().getTopicReference(newTopicName).get().getSubscription("sM")
@@ -398,7 +398,7 @@ public class ClusterMigrationTest {
.subscriptionName("s1-d").subscribe();
Producer<byte[]> producerDiff = client1.newProducer().topic(diffTopic).enableBatching(false)
.producerName("cluster1-d").messageRoutingMode(MessageRoutingMode.SinglePartition).create();
- AbstractTopic topicDiff = (AbstractTopic) pulsar1.getBrokerService().getTopic(diffTopic, false).getNow(null).get();
+ AbstractTopic topicDiff = (AbstractTopic) pulsar2.getBrokerService().getTopic(diffTopic, false).getNow(null).get();
assertNotNull(topicDiff);
for (int i = 0; i < n; i++) {
producerDiff.send("diff".getBytes());
@@ -603,6 +603,39 @@ public class ClusterMigrationTest {
consumer1.close();
producer1.close();
+
+ // publish to new topic which should be redirected immediately
+ String newTopic = topicName+"-new";
+ producer1 = client1.newProducer().topic(newTopic).enableBatching(false)
+ .producerName("cluster1-1").messageRoutingMode(MessageRoutingMode.SinglePartition).create();
+ retryStrategically((test) -> {
+ try {
+ pulsar2.getBrokerService().getTopic(newTopic, false).getNow(null).get();
+ return true;
+ } catch (Exception e) {
+ // ok
+ }
+ return false;
+ }, 10, 500);
+ PersistentTopic pulsar2Topic = (PersistentTopic) pulsar2.getBrokerService().getTopic(newTopic, false).getNow(null)
+ .get();
+ retryStrategically((test) -> {
+ try {
+ return !pulsar2Topic.getProducers().isEmpty();
+ } catch (Exception e) {
+ return false;
+ }
+ }, 10, 500);
+ assertFalse(pulsar2Topic.getProducers().isEmpty());
+ consumer1 = client1.newConsumer().topic(newTopic).subscriptionName("s1").subscribe();
+ retryStrategically((test) -> {
+ try {
+ return !pulsar2Topic.getSubscription("s1").getConsumers().isEmpty();
+ } catch (Exception e) {
+ return false;
+ }
+ }, 10, 500);
+ assertFalse(pulsar2Topic.getSubscription("s1").getConsumers().isEmpty());
}
@Test(dataProvider = "NamespaceMigrationTopicSubscriptionTypes")
@@ -803,7 +836,7 @@ public class ClusterMigrationTest {
consumerM = client1.newConsumer().topic(newTopicName).subscriptionType(subType)
.subscriptionName("sM").subscribe();
retryStrategically((t) -> pulsar1.getBrokerService().getTopicReference(newTopicName).isPresent(), 5, 100);
- pulsar1.getBrokerService().getTopicReference(newTopicName).get().checkClusterMigration().get();
+ pulsar2.getBrokerService().getTopicReference(newTopicName).get().checkClusterMigration().get();
retryStrategically((t) ->
pulsar2.getBrokerService().getTopicReference(newTopicName).isPresent() &&
pulsar2.getBrokerService().getTopicReference(newTopicName).get().getSubscription("sM")
@@ -832,7 +865,7 @@ public class ClusterMigrationTest {
.subscriptionName("s1-d").subscribe();
Producer<byte[]> producerDiff = client1.newProducer().topic(diffTopic).enableBatching(false)
.producerName("cluster1-d").messageRoutingMode(MessageRoutingMode.SinglePartition).create();
- AbstractTopic topicDiff = (AbstractTopic) pulsar1.getBrokerService().getTopic(diffTopic, false).getNow(null).get();
+ AbstractTopic topicDiff = (AbstractTopic) pulsar2.getBrokerService().getTopic(diffTopic, false).getNow(null).get();
assertNotNull(topicDiff);
for (int i = 0; i < n; i++) {
producerDiff.send("diff".getBytes());