rdhabalia commented on a change in pull request #1262: Broker should not start
replicator for root partitioned-topic
URL: https://github.com/apache/incubator-pulsar/pull/1262#discussion_r169782352
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
##########
@@ -583,29 +584,35 @@ private void createPersistentTopic(final String topic,
CompletableFuture<Topic>
new OpenLedgerCallback() {
@Override
public void openLedgerComplete(ManagedLedger ledger,
Object ctx) {
- PersistentTopic persistentTopic = new
PersistentTopic(topic, ledger, BrokerService.this);
-
- CompletableFuture<Void> replicationFuture =
persistentTopic.checkReplication();
- replicationFuture.thenCompose(v -> {
- // Also check dedup status
- return
persistentTopic.checkDeduplicationStatus();
- }).thenRun(() -> {
- log.info("Created topic {} - dedup is {}",
topic,
-
persistentTopic.isDeduplicationEnabled() ? "enabled" : "disabled");
- long topicLoadLatencyMs =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime())
- - topicCreateTimeMs;
- pulsarStats.recordTopicLoadTimeValue(topic,
topicLoadLatencyMs);
- addTopicToStatsMaps(destinationName,
persistentTopic);
- topicFuture.complete(persistentTopic);
- }).exceptionally((ex) -> {
- log.warn("Replication or dedup check failed.
Removing topic from topics list {}, {}", topic, ex);
-
persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
- topics.remove(topic, topicFuture);
- topicFuture.completeExceptionally(ex);
+ try {
+ PersistentTopic persistentTopic = new
PersistentTopic(topic, ledger,
+ BrokerService.this);
+ CompletableFuture<Void> replicationFuture =
persistentTopic.checkReplication();
+ replicationFuture.thenCompose(v -> {
+ // Also check dedup status
+ return
persistentTopic.checkDeduplicationStatus();
+ }).thenRun(() -> {
+ log.info("Created topic {} - dedup is {}",
topic,
+
persistentTopic.isDeduplicationEnabled() ? "enabled" : "disabled");
+ long topicLoadLatencyMs =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime())
+ - topicCreateTimeMs;
+
pulsarStats.recordTopicLoadTimeValue(topic, topicLoadLatencyMs);
+ addTopicToStatsMaps(destinationName,
persistentTopic);
+ topicFuture.complete(persistentTopic);
+ }).exceptionally((ex) -> {
+ log.warn(
+ "Replication or dedup check
failed. Removing topic from topics list {}, {}",
+ topic, ex);
+
persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
+ topics.remove(topic, topicFuture);
+ topicFuture.completeExceptionally(ex);
+ });
+
+ return null;
});
-
- return null;
- });
+ } catch (NamingException e) {
Review comment:
fixed it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
