[feat][broker]PIP-180 Shadow Topic - Part V - Support shadow topic creation.

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -32,6 +32,7 @@
 import org.apache.bookkeeper.common.annotation.InterfaceStability;
 import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
 import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
+import org.apache.commons.collections4.MapUtils;
 import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
 
 /**
@@ -742,4 +743,10 @@ public int getMaxBacklogBetweenCursorsForCaching() {
     public void setMaxBacklogBetweenCursorsForCaching(int maxBacklogBetweenCursorsForCaching) {
         this.maxBacklogBetweenCursorsForCaching = maxBacklogBetweenCursorsForCaching;
     }
+
+    public String getShadowSource() {
+        return MapUtils.getString(properties, PROPERTY_SOURCE_TOPIC_KEY);
+    }
+
+    public static final String PROPERTY_SOURCE_TOPIC_KEY = "PULSAR.SHADOW_SOURCE";
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactory.java

@@ -18,6 +18,7 @@
  */
 package org.apache.bookkeeper.mledger;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
@@ -197,4 +198,8 @@ void asyncOpenReadOnlyCursor(String managedLedgerName, Position startPosition, M
      * */
     long getCacheEvictionTimeThreshold();
 
+    /**
+     * @return properties of this managedLedger.
+     */
+    CompletableFuture<Map<String, String>> getManagedLedgerPropertiesAsync(String name);
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -367,11 +367,13 @@ public void asyncOpen(final String name, final ManagedLedgerConfig config, final
         ledgers.computeIfAbsent(name, (mlName) -> {
             // Create the managed ledger
             CompletableFuture<ManagedLedgerImpl> future = new CompletableFuture<>();
-            final ManagedLedgerImpl newledger = new ManagedLedgerImpl(this,
-                    bookkeeperFactory.get(
-                            new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
-                                    config.getBookKeeperEnsemblePlacementPolicyProperties())),
-                    store, config, scheduledExecutor, name, mlOwnershipChecker);
+            BookKeeper bk = bookkeeperFactory.get(
+                    new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
+                            config.getBookKeeperEnsemblePlacementPolicyProperties()));
+            final ManagedLedgerImpl newledger = config.getShadowSource() == null
+                    ? new ManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name, mlOwnershipChecker)
+                    : new ShadowManagedLedgerImpl(this, bk, store, config, scheduledExecutor, name,
+                    mlOwnershipChecker);
             PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger);
             pendingInitializeLedgers.put(name, pendingLedger);
             newledger.initialize(new ManagedLedgerInitializeLedgerCallback() {
@@ -954,6 +956,11 @@ public void operationFailed(MetaStoreException e) {
         return future;
     }
 
+    @Override
+    public CompletableFuture<Map<String, String>> getManagedLedgerPropertiesAsync(String name) {
+        return store.getManagedLedgerPropertiesAsync(name);
+    }
+
     public MetaStore getMetaStore() {
         return store;
     }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java

@@ -158,4 +158,13 @@ void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorIn
      *         if the operation succeeds.
      */
     CompletableFuture<Boolean> asyncExists(String ledgerName);
+
+
+    /**
+     * Get managed ledger properties from meta store.
+     *
+     * @param name ledgerName
+     * @return a future represents the result of the operation.
+     */
+    CompletableFuture<Map<String, String>> getManagedLedgerPropertiesAsync(String name);
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java

@@ -23,6 +23,8 @@
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -148,6 +150,33 @@ public void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, Map
                 });
     }
 
+    public CompletableFuture<Map<String, String>> getManagedLedgerPropertiesAsync(String name) {
+        CompletableFuture<Map<String, String>> result = new CompletableFuture<>();
+        getManagedLedgerInfo(name, false, new MetaStoreCallback<>() {
+            @Override
+            public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, Stat stat) {
+                HashMap<String, String> propertiesMap = new HashMap<>(mlInfo.getPropertiesCount());
+                if (mlInfo.getPropertiesCount() > 0) {
+                    for (int i = 0; i < mlInfo.getPropertiesCount(); i++) {
+                        MLDataFormats.KeyValue property = mlInfo.getProperties(i);
+                        propertiesMap.put(property.getKey(), property.getValue());
+                    }
+                }
+                result.complete(propertiesMap);
+            }
+
+            @Override
+            public void operationFailed(MetaStoreException e) {
+                if (e instanceof MetadataNotFoundException) {
+                    result.complete(Collections.emptyMap());
+                } else {
+                    result.completeExceptionally(e);
+                }
+            }
+        });
+        return result;
+    }
+
     @Override
     public void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat stat,
             MetaStoreCallback<Void> callback) {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.bookkeeper.mledger.impl;
+
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.pulsar.common.naming.TopicName;
+
+/**
+ * Working in progress until <a href="https://github.com/apache/pulsar/issues/16153">PIP-180</a> is finished.
+ * Currently, it works nothing different with ManagedLedgerImpl.
+ */
+@Slf4j
+public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
+
+    private final TopicName shadowSource;
+    private final String sourceMLName;
+
+    public ShadowManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper,
+                                   MetaStore store, ManagedLedgerConfig config,
+                                   OrderedScheduler scheduledExecutor,
+                                   String name, final Supplier<Boolean> mlOwnershipChecker) {
+        super(factory, bookKeeper, store, config, scheduledExecutor, name, mlOwnershipChecker);
+        this.shadowSource = TopicName.get(config.getShadowSource());
+        this.sourceMLName = shadowSource.getPersistenceNamingEncoding();
+    }
+
+    @Override
+    synchronized void initialize(ManagedLedgerInitializeLedgerCallback callback, Object ctx) {
+        // TODO: ShadowManagedLedger has different initialize process from normal ManagedLedger,
+        //  which is complicated and will be implemented in the next PRs.
+        super.initialize(callback, ctx);
+    }
+
+    public TopicName getShadowSource() {
+        return shadowSource;
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -1408,14 +1408,44 @@ protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(final S
         return topicFuture;
     }
 
+    CompletableFuture<Map<String, String>> fetchTopicPropertiesAsync(TopicName topicName) {
+        if (!topicName.isPartitioned()) {
+            return managedLedgerFactory.getManagedLedgerPropertiesAsync(topicName.getPersistenceNamingEncoding());
+        } else {
+            TopicName partitionedTopicName = TopicName.get(topicName.getPartitionedTopicName());
+            return fetchPartitionedTopicMetadataAsync(partitionedTopicName)
+                    .thenCompose(metadata -> {
+                        if (metadata.partitions == PartitionedTopicMetadata.NON_PARTITIONED) {
+                            return managedLedgerFactory.getManagedLedgerPropertiesAsync(
+                                    topicName.getPersistenceNamingEncoding());
+                        }
+                        return CompletableFuture.completedFuture(metadata.properties);
+                    });
+        }
+    }
+
     private void checkOwnershipAndCreatePersistentTopic(final String topic, boolean createIfMissing,
                                        CompletableFuture<Optional<Topic>> topicFuture,
                                        Map<String, String> properties) {
         TopicName topicName = TopicName.get(topic);
         pulsar.getNamespaceService().isServiceUnitActiveAsync(topicName)
                 .thenAccept(isActive -> {
                     if (isActive) {
-                        createPersistentTopic(topic, createIfMissing, topicFuture, properties);
+                        CompletableFuture<Map<String, String>> propertiesFuture;
+                        if (properties == null) {
+                            //Read properties from storage when loading topic.
+                            propertiesFuture = fetchTopicPropertiesAsync(topicName);
+                        } else {
+                            propertiesFuture = CompletableFuture.completedFuture(properties);
+                        }
+                        propertiesFuture.thenAccept(finalProperties ->
+                                createPersistentTopic(topic, createIfMissing, topicFuture, finalProperties)
+                        ).exceptionally(throwable -> {
+                            log.warn("[{}] Read topic property failed", topic, throwable);
+                            pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
+                            topicFuture.completeExceptionally(throwable);
+                            return null;
+                        });
                     } else {
                         // namespace is being unloaded
                         String msg = String.format("Namespace is being unloaded, cannot add topic %s", topic);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -73,6 +73,7 @@
 import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.impl.ShadowManagedLedgerImpl;
 import org.apache.bookkeeper.net.BookieId;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -177,6 +178,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private final ConcurrentOpenHashMap<String/*ShadowTopic*/, Replicator> shadowReplicators;
     @Getter
     private volatile List<String> shadowTopics;
+    private final TopicName shadowSourceTopic;
 
     static final String DEDUPLICATION_CURSOR_NAME = "pulsar.dedup";
     private static final String TOPIC_EPOCH_PROPERTY_NAME = "pulsar.topic.epoch";
@@ -303,6 +305,11 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS
             this.transactionBuffer = new TransactionBufferDisable();
         }
         transactionBuffer.syncMaxReadPositionForNormalPublish((PositionImpl) ledger.getLastConfirmedEntry());
+        if (ledger instanceof ShadowManagedLedgerImpl) {
+            shadowSourceTopic = ((ShadowManagedLedgerImpl) ledger).getShadowSource();
+        } else {
+            shadowSourceTopic = null;
+        }
     }
 
     @Override
@@ -382,6 +389,7 @@ public CompletableFuture<Void> initialize() {
         } else {
             this.transactionBuffer = new TransactionBufferDisable();
         }
+        shadowSourceTopic = null;
     }
 
     private void initializeDispatchRateLimiterIfNeeded() {
@@ -3328,4 +3336,8 @@ private CompletableFuture<Void> transactionBufferCleanupAndClose() {
     public long getLastDataMessagePublishedTimestamp() {
         return lastDataMessagePublishedTimestamp;
     }
+
+    public Optional<TopicName> getShadowSourceTopic() {
+        return Optional.ofNullable(shadowSourceTopic);
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java

@@ -202,6 +202,8 @@ public void setup() throws Exception {
         doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory();
         doReturn(mock(PulsarClientImpl.class)).when(pulsar).getClient();
 
+        doAnswer(invocationOnMock -> CompletableFuture.completedFuture(null))
+                .when(mlFactoryMock).getManagedLedgerPropertiesAsync(any());
         doAnswer(invocation -> {
             DeleteLedgerCallback deleteLedgerCallback = invocation.getArgument(1);
             deleteLedgerCallback.deleteLedgerComplete(null);

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.broker.service.persistent;
+
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.impl.ShadowManagedLedgerImpl;
+import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.common.naming.TopicName;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class ShadowTopicTest extends BrokerTestBase {
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        baseSetup();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        internalCleanup();
+    }
+
+    @Test()
+    public void testNonPartitionedShadowTopicSetup() throws Exception {
+        String sourceTopic = "persistent://prop/ns-abc/source";
+        String shadowTopic = "persistent://prop/ns-abc/shadow";
+        //1. test shadow topic setting in topic creation.
+        admin.topics().createNonPartitionedTopic(sourceTopic);
+        admin.topics().createShadowTopic(shadowTopic, sourceTopic);
+        PersistentTopic brokerShadowTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopic).get().get();
+        Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
+        Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
+        Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
+
+        //2. test shadow topic could be properly loaded after unload.
+        admin.namespaces().unload("prop/ns-abc");
+        Assert.assertTrue(pulsar.getBrokerService().getTopicReference(shadowTopic).isEmpty());
+        Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
+        brokerShadowTopic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopic).get().get();
+        Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
+        Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
+    }
+
+    @Test()
+    public void testPartitionedShadowTopicSetup() throws Exception {
+        String sourceTopic = "persistent://prop/ns-abc/source-p";
+        String shadowTopic = "persistent://prop/ns-abc/shadow-p";
+        String shadowTopicPartition = TopicName.get(shadowTopic).getPartition(0).toString();
+
+        //1. test shadow topic setting in topic creation.
+        admin.topics().createPartitionedTopic(sourceTopic, 2);
+        admin.topics().createShadowTopic(shadowTopic, sourceTopic);
+        pulsarClient.newProducer().topic(shadowTopic).create().close();//trigger loading partitions.
+        PersistentTopic brokerShadowTopic = (PersistentTopic) pulsar.getBrokerService()
+                .getTopicIfExists(shadowTopicPartition).get().get();
+        Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
+        Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
+        Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
+
+        //2. test shadow topic could be properly loaded after unload.
+        admin.namespaces().unload("prop/ns-abc");
+        Assert.assertTrue(pulsar.getBrokerService().getTopicReference(shadowTopic).isEmpty());
+
+        Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
+        brokerShadowTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopicPartition).get().get();
+        Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
+        Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
+    }
+
+
+}