PIP 110: Support Topic metadata - PART-1 create topic with properties

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

@@ -77,6 +77,7 @@ public class ManagedLedgerConfig {
     private int newEntriesCheckDelayInMillis = 10;
     private Clock clock = Clock.systemUTC();
     private ManagedLedgerInterceptor managedLedgerInterceptor;
+    private Map<String, String> topicMetadata;
 
     public boolean isCreateIfMissing() {
         return createIfMissing;
@@ -623,6 +624,16 @@ public void setBookKeeperEnsemblePlacementPolicyProperties(
         this.bookKeeperEnsemblePlacementPolicyProperties = bookKeeperEnsemblePlacementPolicyProperties;
     }
 
+
+    public Map<String, String> getTopicMetadata() {
+        return topicMetadata;
+    }
+
+
+    public void setTopicMetadata(Map<String, String> topicMetadata) {
+        this.topicMetadata = topicMetadata;
+    }
+
     public boolean isDeletionAtBatchIndexLevelEnabled() {
         return deletionAtBatchIndexLevelEnabled;
     }

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

@@ -325,7 +325,8 @@ synchronized void initialize(final ManagedLedgerInitializeLedgerCallback callbac
         log.info("Opening managed ledger {}", name);
 
         // Fetch the list of existing ledgers in the managed ledger
-        store.getManagedLedgerInfo(name, config.isCreateIfMissing(), new MetaStoreCallback<ManagedLedgerInfo>() {
+        store.getManagedLedgerInfo(name, config.isCreateIfMissing(), config.getTopicMetadata(),
+                new MetaStoreCallback<ManagedLedgerInfo>() {
             @Override
             public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
                 ledgersStat = stat;

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

@@ -19,6 +19,7 @@
 package org.apache.bookkeeper.mledger.impl;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
@@ -52,7 +53,24 @@ interface MetaStoreCallback<T> {
      *            whether the managed ledger metadata should be created if it doesn't exist already
      * @throws MetaStoreException
      */
-    void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, MetaStoreCallback<ManagedLedgerInfo> callback);
+    default void getManagedLedgerInfo(String ledgerName, boolean createIfMissing,
+                              MetaStoreCallback<ManagedLedgerInfo> callback) {
+        getManagedLedgerInfo(ledgerName, createIfMissing, null, callback);
+    }
+
+    /**
+     * Get the metadata used by the ManagedLedger.
+     *
+     * @param ledgerName
+     *            the name of the ManagedLedger
+     * @param createIfMissing
+     *            whether the managed ledger metadata should be created if it doesn't exist already
+     * @param topicMetadata
+     *            topic metadata
+     * @throws MetaStoreException
+     */
+    void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, Map<String, String> topicMetadata,
+                              MetaStoreCallback<ManagedLedgerInfo> callback);
 
     /**
      *

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

@@ -22,6 +22,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -84,7 +85,7 @@ public MetaStoreImpl(MetadataStore store, OrderedExecutor executor, String compr
     }
 
     @Override
-    public void getManagedLedgerInfo(String ledgerName, boolean createIfMissing,
+    public void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, Map<String, String> topicMetadata,
             MetaStoreCallback<ManagedLedgerInfo> callback) {
         // Try to get the content or create an empty node
         String path = PREFIX + ledgerName;
@@ -106,8 +107,17 @@ public void getManagedLedgerInfo(String ledgerName, boolean createIfMissing,
 
                             store.put(path, new byte[0], Optional.of(-1L))
                                     .thenAccept(stat -> {
-                                        ManagedLedgerInfo info = ManagedLedgerInfo.getDefaultInstance();
-                                        callback.operationComplete(info, stat);
+                                        ManagedLedgerInfo.Builder ledgerBuilder = ManagedLedgerInfo.newBuilder();
+                                        if (topicMetadata != null) {
+                                            topicMetadata.forEach((k, v) -> {
+                                                ledgerBuilder.addProperties(
+                                                        MLDataFormats.KeyValue.newBuilder()
+                                                                .setKey(k)
+                                                                .setValue(v)
+                                                                .build());
+                                            });
+                                        }
+                                        callback.operationComplete(ledgerBuilder.build(), stat);
                                     }).exceptionally(ex -> {
                                         callback.operationFailed(getException(ex));
                                         return null;

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -1632,6 +1632,36 @@ public void cursorReadsWithDiscardedEmptyLedgers() throws Exception {
         assertEquals(c1.readEntries(1).size(), 0);
     }
 
+    @Test
+    public void testSetTopicMetadata() throws Exception {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("key1", "value1");
+        properties.put("key2", "value2");
+        final MetaStore store = factory.getMetaStore();
+        final CountDownLatch latch = new CountDownLatch(1);
+        final ManagedLedgerInfo[] storedMLInfo = new ManagedLedgerInfo[1];
+        store.getManagedLedgerInfo("my_test_ledger", true, properties, new MetaStoreCallback<ManagedLedgerInfo>() {
+            @Override
+            public void operationComplete(ManagedLedgerInfo result, Stat version) {
+                storedMLInfo[0] = result;
+                latch.countDown();
+            }
+
+            @Override
+            public void operationFailed(MetaStoreException e) {
+                latch.countDown();
+                fail("Should have failed here");
+            }
+        });
+        latch.await();
+
+        assertEquals(storedMLInfo[0].getPropertiesCount(), 2);
+        assertEquals(storedMLInfo[0].getPropertiesList().get(0).getKey(), "key1");
+        assertEquals(storedMLInfo[0].getPropertiesList().get(0).getValue(), "value1");
+        assertEquals(storedMLInfo[0].getPropertiesList().get(1).getKey(), "key2");
+        assertEquals(storedMLInfo[0].getPropertiesList().get(1).getValue(), "value2");
+    }
+
     @Test
     public void cursorReadsWithDiscardedEmptyLedgersStillListed() throws Exception {
         ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("my_test_ledger");

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -22,6 +22,7 @@
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -574,6 +575,11 @@ protected List<String> getTopicPartitionList(TopicDomain topicDomain) {
 
     protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions,
                                                   boolean createLocalTopicOnly) {
+        internalCreatePartitionedTopic(asyncResponse, numPartitions, createLocalTopicOnly, null);
+    }
+
+    protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions,
+                                                  boolean createLocalTopicOnly, Map<String, String> topicMetadata) {
         Integer maxTopicsPerNamespace = null;
 
         try {
@@ -640,7 +646,7 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
                 return;
             }
 
-            provisionPartitionedTopicPath(asyncResponse, numPartitions, createLocalTopicOnly)
+            provisionPartitionedTopicPath(asyncResponse, numPartitions, createLocalTopicOnly, topicMetadata)
                     .thenCompose(ignored -> tryCreatePartitionsAsync(numPartitions))
                     .whenComplete((ignored, ex) -> {
                         if (ex != null) {
@@ -679,7 +685,7 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
                                 ((TopicsImpl) pulsar().getBrokerService()
                                         .getClusterPulsarAdmin(cluster, clusterDataOp).topics())
                                         .createPartitionedTopicAsync(
-                                                topicName.getPartitionedTopicName(), numPartitions, true);
+                                                topicName.getPartitionedTopicName(), numPartitions, true, null);
                             })
                             .exceptionally(throwable -> {
                                 log.error("Failed to create partition topic in cluster {}.", cluster, throwable);
@@ -718,13 +724,13 @@ protected CompletableFuture<Boolean> checkTopicExistsAsync(TopicName topicName)
                 });
     }
 
-    private CompletableFuture<Void> provisionPartitionedTopicPath(AsyncResponse asyncResponse,
-                                                                  int numPartitions,
-                                                                  boolean createLocalTopicOnly) {
+    private CompletableFuture<Void> provisionPartitionedTopicPath(AsyncResponse asyncResponse, int numPartitions,
+                                                                  boolean createLocalTopicOnly,
+                                                                  Map<String, String> topicMetadata) {
         CompletableFuture<Void> future = new CompletableFuture<>();
         namespaceResources()
                 .getPartitionedTopicResources()
-                .createPartitionedTopicAsync(topicName, new PartitionedTopicMetadata(numPartitions))
+                .createPartitionedTopicAsync(topicName, new PartitionedTopicMetadata(numPartitions, topicMetadata))
                 .whenComplete((ignored, ex) -> {
                     if (ex != null) {
                         if (ex instanceof AlreadyExistsException) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -357,7 +357,7 @@ protected void internalRevokePermissionsOnTopic(String role) {
         revokePermissions(topicName.toString(), role);
     }
 
-    protected void internalCreateNonPartitionedTopic(boolean authoritative) {
+    protected void internalCreateNonPartitionedTopic(boolean authoritative, Map<String, String> topicMetadata) {
         validateNonPartitionTopicName(topicName.getLocalName());
         if (topicName.isGlobal()) {
             validateGlobalNamespaceOwnership(namespaceName);
@@ -378,7 +378,7 @@ protected void internalCreateNonPartitionedTopic(boolean authoritative) {
                 throw new RestException(Status.CONFLICT, "This topic already exists");
             }
 
-            Topic createdTopic = getOrCreateTopic(topicName);
+            Topic createdTopic = getOrCreateTopic(topicName, topicMetadata);
             log.info("[{}] Successfully created non-partitioned topic {}", clientAppId(), createdTopic);
         } catch (Exception e) {
             if (e instanceof RestException) {
@@ -3750,8 +3750,12 @@ private CompletableFuture<Topic> topicNotFoundReasonAsync(TopicName topicName) {
     }
 
     private Topic getOrCreateTopic(TopicName topicName) {
-        return pulsar().getBrokerService().getTopic(
-                topicName.toString(), true).thenApply(Optional::get).join();
+        return getOrCreateTopic(topicName, null);
+    }
+
+    private Topic getOrCreateTopic(TopicName topicName, Map<String, String> topicMetadata) {
+        return pulsar().getBrokerService().getTopic(topicName.toString(), true, topicMetadata)
+                .thenApply(Optional::get).join();
     }
 
     /**

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java

@@ -171,6 +171,34 @@ public void createPartitionedTopic(
         }
     }
 
+    @PUT
+    @Path("/{property}/{cluster}/{namespace}/{topic}/partitions/topicMetadata")
+    @ApiOperation(hidden = true, value = "Create a partitioned topic.",
+            notes = "It needs to be called before creating a producer on a partitioned topic.")
+    @ApiResponses(value = {
+            @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
+            @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 406, message = "The number of partitions should be "
+                    + "more than 0 and less than or equal to maxNumPartitionsPerPartitionedTopic"),
+            @ApiResponse(code = 409, message = "Partitioned topic already exist")})
+    public void createPartitionedTopic(
+            @Suspended final AsyncResponse asyncResponse,
+            @PathParam("property") String property,
+            @PathParam("cluster") String cluster,
+            @PathParam("namespace") String namespace,
+            @PathParam("topic") @Encoded String encodedTopic,
+            @QueryParam("createLocalTopicOnly") @DefaultValue("false") boolean createLocalTopicOnly,
+            PartitionedTopicMetadata metadata) {
+        try {
+            validateTopicName(property, cluster, namespace, encodedTopic);
+            internalCreatePartitionedTopic(asyncResponse, metadata.partitions, createLocalTopicOnly,
+                    metadata.topicMetadata);
+        } catch (Exception e) {
+            log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+            resumeAsyncResponseExceptionally(asyncResponse, e);
+        }
+    }
+
     @PUT
     @Path("/{tenant}/{cluster}/{namespace}/{topic}")
     @ApiOperation(value = "Create a non-partitioned topic.",
@@ -196,10 +224,39 @@ public void createNonPartitionedTopic(
             @PathParam("topic") @Encoded String encodedTopic,
             @ApiParam(value = "Is authentication required to perform this operation")
             @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+        createNonPartitionedTopic(tenant, cluster, namespace, encodedTopic, authoritative, null);
+    }
+
+    @PUT
+    @Path("/{tenant}/{cluster}/{namespace}/{topic}/topicMetadata")
+    @ApiOperation(value = "Create a non-partitioned topic.",
+            notes = "This is the only REST endpoint from which non-partitioned topics could be created.")
+    @ApiResponses(value = {
+            @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
+            @ApiResponse(code = 401, message = "Don't have permission to administrate resources on this tenant"),
+            @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 409, message = "Partitioned topic already exist"),
+            @ApiResponse(code = 412,
+                    message = "Failed Reason : Name is invalid or Namespace does not have any clusters configured"),
+            @ApiResponse(code = 500, message = "Internal server error"),
+            @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
+    })
+    public void createNonPartitionedTopic(
+            @ApiParam(value = "Specify the tenant", required = true)
+            @PathParam("tenant") String tenant,
+            @ApiParam(value = "Specify the cluster", required = true)
+            @PathParam("cluster") String cluster,
+            @ApiParam(value = "Specify the namespace", required = true)
+            @PathParam("namespace") String namespace,
+            @ApiParam(value = "Specify topic name", required = true)
+            @PathParam("topic") @Encoded String encodedTopic,
+            @ApiParam(value = "Is authentication required to perform this operation")
+            @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+            @ApiParam(value = "Key value pair properties for the topic metadata") Map<String, String> topicMetadata) {
         validateNamespaceName(tenant, cluster, namespace);
         validateTopicName(tenant, cluster, namespace, encodedTopic);
         validateGlobalNamespaceOwnership();
-        internalCreateNonPartitionedTopic(authoritative);
+        internalCreateNonPartitionedTopic(authoritative, topicMetadata);
     }
 
     /**