diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java index e18e337d41881..4230c01b3c593 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java @@ -73,6 +73,7 @@ public class ManagedLedgerConfig { private int newEntriesCheckDelayInMillis = 10; private Clock clock = Clock.systemUTC(); private ManagedLedgerInterceptor managedLedgerInterceptor; + private Map properties; public boolean isCreateIfMissing() { return createIfMissing; @@ -619,6 +620,16 @@ public void setBookKeeperEnsemblePlacementPolicyProperties( this.bookKeeperEnsemblePlacementPolicyProperties = bookKeeperEnsemblePlacementPolicyProperties; } + + public Map getProperties() { + return properties; + } + + + public void setProperties(Map properties) { + this.properties = properties; + } + public boolean isDeletionAtBatchIndexLevelEnabled() { return deletionAtBatchIndexLevelEnabled; } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index a035e0e4786a9..1d24c387f7516 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -329,7 +329,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() { + store.getManagedLedgerInfo(name, config.isCreateIfMissing(), config.getProperties(), + new MetaStoreCallback() { @Override public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) { ledgersStat = stat; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java index aca8e4efac6a4..35f109b21dc5c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java +++ b/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; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; @@ -51,7 +52,23 @@ interface MetaStoreCallback { * whether the managed ledger metadata should be created if it doesn't exist already * @throws MetaStoreException */ - void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, + default void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, + MetaStoreCallback 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 properties + * ledger properties + * @throws MetaStoreException + */ + void getManagedLedgerInfo(String ledgerName, boolean createIfMissing, Map properties, MetaStoreCallback callback); /** diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java index ac2e746ac4676..1c03c481ba482 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java @@ -24,6 +24,7 @@ import io.netty.buffer.Unpooled; 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; @@ -81,7 +82,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 properties, MetaStoreCallback callback) { // Try to get the content or create an empty node String path = PREFIX + ledgerName; @@ -103,8 +104,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 (properties != null) { + properties.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; diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 79eab77de1f88..95ab0024bf9e2 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/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 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() { + @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"); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index f802858a47889..e5d533200c971 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/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; @@ -568,6 +569,11 @@ protected List 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 properties) { Integer maxTopicsPerNamespace = null; try { @@ -634,7 +640,7 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n return; } - provisionPartitionedTopicPath(asyncResponse, numPartitions, createLocalTopicOnly) + provisionPartitionedTopicPath(asyncResponse, numPartitions, createLocalTopicOnly, properties) .thenCompose(ignored -> tryCreatePartitionsAsync(numPartitions)) .whenComplete((ignored, ex) -> { if (ex != null) { @@ -673,7 +679,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); @@ -712,13 +718,13 @@ protected CompletableFuture checkTopicExistsAsync(TopicName topicName) }); } - private CompletableFuture provisionPartitionedTopicPath(AsyncResponse asyncResponse, - int numPartitions, - boolean createLocalTopicOnly) { + private CompletableFuture provisionPartitionedTopicPath(AsyncResponse asyncResponse, int numPartitions, + boolean createLocalTopicOnly, + Map properties) { CompletableFuture future = new CompletableFuture<>(); namespaceResources() .getPartitionedTopicResources() - .createPartitionedTopicAsync(topicName, new PartitionedTopicMetadata(numPartitions)) + .createPartitionedTopicAsync(topicName, new PartitionedTopicMetadata(numPartitions, properties)) .whenComplete((ignored, ex) -> { if (ex != null) { if (ex instanceof AlreadyExistsException) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 5e651cf892b3d..05546453c231f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -356,7 +356,7 @@ protected void internalRevokePermissionsOnTopic(String role) { revokePermissions(topicName.toString(), role); } - protected void internalCreateNonPartitionedTopic(boolean authoritative) { + protected void internalCreateNonPartitionedTopic(boolean authoritative, Map properties) { validateNonPartitionTopicName(topicName.getLocalName()); if (topicName.isGlobal()) { validateGlobalNamespaceOwnership(namespaceName); @@ -377,7 +377,7 @@ protected void internalCreateNonPartitionedTopic(boolean authoritative) { throw new RestException(Status.CONFLICT, "This topic already exists"); } - Topic createdTopic = getOrCreateTopic(topicName); + Topic createdTopic = getOrCreateTopic(topicName, properties); log.info("[{}] Successfully created non-partitioned topic {}", clientAppId(), createdTopic); } catch (Exception e) { if (e instanceof RestException) { @@ -3811,8 +3811,12 @@ private CompletableFuture 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 properties) { + return pulsar().getBrokerService().getTopic(topicName.toString(), true, properties) + .thenApply(Optional::get).join(); } /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java index 1c5ef250ccd42..92c791da4bf96 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java @@ -201,7 +201,7 @@ public void createNonPartitionedTopic( validateNamespaceName(tenant, cluster, namespace); validateTopicName(tenant, cluster, namespace, encodedTopic); validateGlobalNamespaceOwnership(); - internalCreateNonPartitionedTopic(authoritative); + internalCreateNonPartitionedTopic(authoritative, null); } /** diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index 2bc058a1ae4c8..4f9231429995f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java @@ -264,12 +264,14 @@ public void createNonPartitionedTopic( @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) { + @QueryParam("authoritative") @DefaultValue("false") boolean authoritative, + @ApiParam(value = "Key value pair properties for the topic metadata") + Map properties) { validateNamespaceName(tenant, namespace); validateGlobalNamespaceOwnership(); validateTopicName(tenant, namespace, encodedTopic); validateCreateTopic(topicName); - internalCreateNonPartitionedTopic(authoritative); + internalCreateNonPartitionedTopic(authoritative, properties); } @GET diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/PersistentTopics.java new file mode 100644 index 0000000000000..f7960d8392cc0 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/PersistentTopics.java @@ -0,0 +1,93 @@ +/** + * 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.admin.v3; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.Encoded; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.container.AsyncResponse; +import javax.ws.rs.container.Suspended; +import javax.ws.rs.core.MediaType; +import org.apache.pulsar.broker.admin.impl.PersistentTopicsBase; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; +import org.apache.pulsar.common.policies.data.PolicyName; +import org.apache.pulsar.common.policies.data.PolicyOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + */ +@Path("/persistent") +@Produces(MediaType.APPLICATION_JSON) +@Api(value = "/persistent", description = "Persistent topic admin apis", tags = "persistent topic") +public class PersistentTopics extends PersistentTopicsBase { + + @PUT + @Path("/{tenant}/{namespace}/{topic}/partitions") + @ApiOperation(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 = 401, message = "Don't have permission to administrate resources on this tenant"), + @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Tenant does not exist"), + @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"), + @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 createPartitionedTopic( + @Suspended final AsyncResponse asyncResponse, + @ApiParam(value = "Specify the tenant", required = true) + @PathParam("tenant") String tenant, + @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 = "The metadata for the topic", + required = true, type = "PartitionedTopicMetadata") PartitionedTopicMetadata metadata, + @QueryParam("createLocalTopicOnly") @DefaultValue("false") boolean createLocalTopicOnly) { + try { + validateNamespaceName(tenant, namespace); + validateGlobalNamespaceOwnership(); + validatePartitionedTopicName(tenant, namespace, encodedTopic); + validateTopicPolicyOperation(topicName, PolicyName.PARTITION, PolicyOperation.WRITE); + validateCreateTopic(topicName); + internalCreatePartitionedTopic(asyncResponse, metadata.partitions, createLocalTopicOnly, + metadata.properties); + } catch (Exception e) { + log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e); + resumeAsyncResponseExceptionally(asyncResponse, e); + } + } + + private static final Logger log = LoggerFactory.getLogger(PersistentTopics.class); +} 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 c2a27ad37fc97..cfe67a44196c5 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 @@ -75,6 +75,7 @@ import java.util.function.Predicate; import javax.ws.rs.core.Response; import lombok.AccessLevel; +import lombok.AllArgsConstructor; import lombok.Getter; import lombok.Setter; import org.apache.bookkeeper.common.util.OrderedExecutor; @@ -89,8 +90,6 @@ import org.apache.bookkeeper.mledger.ManagedLedgerFactory; import org.apache.bookkeeper.mledger.util.Futures; import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.tuple.ImmutablePair; -import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.bookie.rackawareness.IsolatedBookieEnsemblePlacementPolicy; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; @@ -215,7 +214,7 @@ public class BrokerService implements Closeable { prepareDynamicConfigurationMap(); private final ConcurrentOpenHashMap> configRegisteredListeners; - private final ConcurrentLinkedQueue>>> pendingTopicLoadingQueue; + private final ConcurrentLinkedQueue pendingTopicLoadingQueue; private AuthorizationService authorizationService = null; private final ScheduledExecutorService statsUpdater; @@ -905,6 +904,11 @@ public CompletableFuture getOrCreateTopic(final String topic) { } public CompletableFuture> getTopic(final String topic, boolean createIfMissing) { + return getTopic(topic, createIfMissing, null); + } + + public CompletableFuture> getTopic(final String topic, boolean createIfMissing, + Map properties) { try { CompletableFuture> topicFuture = topics.get(topic); if (topicFuture != null) { @@ -921,7 +925,7 @@ public CompletableFuture> getTopic(final String topic, boolean c return topicFuture.thenCompose(value -> { if (!value.isPresent()) { // retry and create topic - return getTopic(topic, createIfMissing); + return getTopic(topic, createIfMissing, properties); } else { // in-progress future completed successfully return CompletableFuture.completedFuture(value); @@ -936,7 +940,7 @@ public CompletableFuture> getTopic(final String topic, boolean c final boolean isPersistentTopic = TopicName.get(topic).getDomain().equals(TopicDomain.persistent); if (isPersistentTopic) { return topics.computeIfAbsent(topic, (topicName) -> { - return this.loadOrCreatePersistentTopic(topicName, createIfMissing); + return this.loadOrCreatePersistentTopic(topicName, createIfMissing, properties); }); } else { return topics.computeIfAbsent(topic, (name) -> { @@ -1279,7 +1283,7 @@ public PulsarAdmin getClusterPulsarAdmin(String cluster, Optional c * @throws RuntimeException */ protected CompletableFuture> loadOrCreatePersistentTopic(final String topic, - boolean createIfMissing) throws RuntimeException { + boolean createIfMissing, Map properties) throws RuntimeException { final CompletableFuture> topicFuture = FutureUtil.createFutureWithTimeout( Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(), () -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION); @@ -1297,7 +1301,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get(); if (topicLoadSemaphore.tryAcquire()) { - createPersistentTopic(topic, createIfMissing, topicFuture); + createPersistentTopic(topic, createIfMissing, topicFuture, properties); topicFuture.handle((persistentTopic, ex) -> { // release permit and process pending topic topicLoadSemaphore.release(); @@ -1305,7 +1309,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S return null; }); } else { - pendingTopicLoadingQueue.add(new ImmutablePair<>(topic, topicFuture)); + pendingTopicLoadingQueue.add(new TopicLoadingContext(topic, topicFuture, properties)); if (log.isDebugEnabled()) { log.debug("topic-loading for {} added into pending queue", topic); } @@ -1319,7 +1323,8 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S } private void createPersistentTopic(final String topic, boolean createIfMissing, - CompletableFuture> topicFuture) { + CompletableFuture> topicFuture, + Map properties) { final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); TopicName topicName = TopicName.get(topic); @@ -1360,6 +1365,7 @@ private void createPersistentTopic(final String topic, boolean createIfMissing, new ManagedLedgerInterceptorImpl(interceptors, brokerEntryPayloadProcessors)); } managedLedgerConfig.setCreateIfMissing(createIfMissing); + managedLedgerConfig.setProperties(properties); // Once we have the configuration, we can proceed with the async open operation managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig, @@ -1376,7 +1382,6 @@ public void openLedgerComplete(ManagedLedger ledger, Object ctx) { .initialize() .thenCompose(__ -> persistentTopic.checkReplication()); - CompletableFuture.allOf(preCreateSubForCompaction, replicationFuture) .thenCompose(v -> { // Also check dedup status @@ -2398,17 +2403,17 @@ private ConcurrentOpenHashMap getRuntimeConfigurationMap() { * permit if it was successful to acquire it. */ private void createPendingLoadTopic() { - Pair>> pendingTopic = pendingTopicLoadingQueue.poll(); + TopicLoadingContext pendingTopic = pendingTopicLoadingQueue.poll(); if (pendingTopic == null) { return; } - final String topic = pendingTopic.getLeft(); + final String topic = pendingTopic.getTopic(); checkTopicNsOwnership(topic).thenRun(() -> { - CompletableFuture> pendingFuture = pendingTopic.getRight(); + CompletableFuture> pendingFuture = pendingTopic.getTopicFuture(); final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get(); final boolean acquiredPermit = topicLoadSemaphore.tryAcquire(); - createPersistentTopic(topic, true, pendingFuture); + createPersistentTopic(topic, true, pendingFuture, pendingTopic.getProperties()); pendingFuture.handle((persistentTopic, ex) -> { // release permit and process next pending topic if (acquiredPermit) { @@ -2419,7 +2424,7 @@ private void createPendingLoadTopic() { }); }).exceptionally(e -> { log.error("Failed to create pending topic {}", topic, e); - pendingTopic.getRight() + pendingTopic.getTopicFuture() .completeExceptionally((e instanceof RuntimeException && e.getCause() != null) ? e.getCause() : e); // schedule to process next pending topic inactivityMonitor.schedule(this::createPendingLoadTopic, 100, TimeUnit.MILLISECONDS); @@ -2821,4 +2826,12 @@ public long getPausedConnections() { public void setPulsarChannelInitializerFactory(PulsarChannelInitializer.Factory factory) { this.pulsarChannelInitFactory = factory; } + + @AllArgsConstructor + @Getter + private static class TopicLoadingContext { + private final String topic; + private final CompletableFuture> topicFuture; + private final Map properties; + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index 37d10854db8bf..4be224490ff78 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -87,6 +87,7 @@ import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.zookeeper.KeeperException; +import org.awaitility.Awaitility; import org.mockito.ArgumentCaptor; import org.powermock.reflect.Whitebox; import org.testng.Assert; @@ -94,12 +95,14 @@ import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import org.testng.collections.Maps; @Slf4j @Test(groups = "broker-admin") public class PersistentTopicsTest extends MockedPulsarServiceBaseTest { private PersistentTopics persistentTopics; + private org.apache.pulsar.broker.admin.v3.PersistentTopics persistentTopicsV3; private final String testTenant = "my-tenant"; private final String testLocalCluster = "use"; private final String testNamespace = "my-namespace"; @@ -121,6 +124,9 @@ protected void setup() throws Exception { persistentTopics = spy(PersistentTopics.class); persistentTopics.setServletContext(new MockServletContext()); persistentTopics.setPulsar(pulsar); + persistentTopicsV3 = spy(org.apache.pulsar.broker.admin.v3.PersistentTopics.class); + persistentTopicsV3.setServletContext(new MockServletContext()); + persistentTopicsV3.setPulsar(pulsar); doReturn(false).when(persistentTopics).isRequestHttps(); doReturn(null).when(persistentTopics).originalPrincipal(); doReturn("test").when(persistentTopics).clientAppId(); @@ -128,6 +134,13 @@ protected void setup() throws Exception { doNothing().when(persistentTopics).validateAdminAccessForTenant(this.testTenant); doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopics).clientAuthData(); + doReturn(false).when(persistentTopicsV3).isRequestHttps(); + doReturn(null).when(persistentTopicsV3).originalPrincipal(); + doReturn("test").when(persistentTopicsV3).clientAppId(); + doReturn(TopicDomain.persistent.value()).when(persistentTopicsV3).domain(); + doNothing().when(persistentTopicsV3).validateAdminAccessForTenant(this.testTenant); + doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopicsV3).clientAuthData(); + nonPersistentTopic = spy(NonPersistentTopics.class); nonPersistentTopic.setServletContext(new MockServletContext()); nonPersistentTopic.setPulsar(pulsar); @@ -349,7 +362,7 @@ public void testTerminate() { String testLocalTopicName = "topic-not-found"; // 1) Create the nonPartitionTopic topic - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, testLocalTopicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, testLocalTopicName, true, null); // 2) Create a subscription AsyncResponse response = mock(AsyncResponse.class); @@ -391,7 +404,7 @@ public void testNonPartitionedTopics() { Assert.assertTrue(errorCaptor.getValue().getMessage().contains("zero partitions")); final String nonPartitionTopic2 = "secondary-non-partitioned-topic"; - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, nonPartitionTopic2, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, nonPartitionTopic2, true, null); Assert.assertEquals(persistentTopics .getPartitionedMetadata(testTenant, testNamespace, nonPartitionTopic, true, false).partitions, 0); @@ -404,7 +417,7 @@ public void testNonPartitionedTopics() { @Test public void testCreateNonPartitionedTopic() { final String topicName = "standard-topic-partition-a"; - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); PartitionedTopicMetadata pMetadata = persistentTopics.getPartitionedMetadata( testTenant, testNamespace, topicName, true, false); Assert.assertEquals(pMetadata.partitions, 0); @@ -412,6 +425,36 @@ public void testCreateNonPartitionedTopic() { PartitionedTopicMetadata metadata = persistentTopics.getPartitionedMetadata( testTenant, testNamespace, topicName, true, true); Assert.assertEquals(metadata.partitions, 0); + final String topicName2 = "standard-topic-partition-b"; + Map topicMetadata = Maps.newHashMap(); + topicMetadata.put("key1", "value1"); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName2, true, topicMetadata); + PartitionedTopicMetadata pMetadata2 = persistentTopics.getPartitionedMetadata( + testTenant, testNamespace, topicName2, true, false); + Assert.assertNull(pMetadata2.properties); + } + + @Test + public void testCreatePartitionedTopic() { + AsyncResponse response = mock(AsyncResponse.class); + final String topicName = "standard-partitioned-topic-a"; + persistentTopics.createPartitionedTopic(response, testTenant, testNamespace, topicName, 2, true); + Awaitility.await().untilAsserted(() -> { + PartitionedTopicMetadata pMetadata = persistentTopics.getPartitionedMetadata( + testTenant, testNamespace, topicName, true, false); + Assert.assertNull(pMetadata.properties); + }); + final String topicName2 = "standard-partitioned-topic-b"; + Map topicMetadata = Maps.newHashMap(); + topicMetadata.put("key1", "value1"); + PartitionedTopicMetadata metadata = new PartitionedTopicMetadata(2, topicMetadata); + persistentTopicsV3.createPartitionedTopic(response, testTenant, testNamespace, topicName2, metadata, true); + Awaitility.await().untilAsserted(() -> { + PartitionedTopicMetadata pMetadata2 = persistentTopics.getPartitionedMetadata( + testTenant, testNamespace, topicName2, true, false); + Assert.assertEquals(pMetadata2.properties.size(), 1); + Assert.assertEquals(pMetadata2.properties, topicMetadata); + }); } @Test(expectedExceptions = RestException.class) @@ -422,7 +465,7 @@ public void testCreateNonPartitionedTopicWithInvalidName() { assert(partitionedTopicname.getLocalName().equals("standard-topic")); return new PartitionedTopicMetadata(10); }).when(persistentTopics).getPartitionedTopicMetadata(any(), anyBoolean(), anyBoolean()); - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); } @Test @@ -484,7 +527,7 @@ public void testUnloadTopic() { // 2) create non partitioned topic and unload response = mock(AsyncResponse.class); - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); persistentTopics.unloadTopic(response, testTenant, testNamespace, topicName, true); ArgumentCaptor responseCaptor = ArgumentCaptor.forClass(Response.class); verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); @@ -546,7 +589,7 @@ public void testGetPartitionedTopicsList() throws KeeperException, InterruptedEx @Test public void testGrantNonPartitionedTopic() { final String topicName = "non-partitioned-topic"; - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); String role = "role"; Set expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); @@ -563,7 +606,7 @@ public void testCreateExistedPartition() { final String partitionName = TopicName.get(topicName).getPartition(0).getLocalName(); try { - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, partitionName, false); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, partitionName, false, null); Assert.fail(); } catch (RestException e) { log.error("Failed to create {}: {}", partitionName, e.getMessage()); @@ -603,7 +646,7 @@ public void testGrantPartitionedTopic() { @Test public void testRevokeNonPartitionedTopic() { final String topicName = "non-partitioned-topic"; - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); String role = "role"; Set expectActions = new HashSet<>(); expectActions.add(AuthAction.produce); @@ -655,7 +698,7 @@ public void testTriggerCompactionTopic() { // create non partitioned topic and compaction on it response = mock(AsyncResponse.class); - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, nonPartitionTopicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, nonPartitionTopicName, true, null); persistentTopics.compact(response, testTenant, testNamespace, nonPartitionTopicName, true); ArgumentCaptor responseCaptor = ArgumentCaptor.forClass(Response.class); verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); @@ -679,7 +722,7 @@ public void testPeekWithSubscriptionNameNotExist() throws Exception { admin.namespaces().setRetention("tenant-xyz/ns-abc", retention); final String topic = "persistent://tenant-xyz/ns-abc/topic-testPeekWithSubscriptionNameNotExist"; final String subscriptionName = "sub"; - ((TopicsImpl) admin.topics()).createPartitionedTopicAsync(topic, 3, true).get(); + ((TopicsImpl) admin.topics()).createPartitionedTopicAsync(topic, 3, true, null).get(); final String partitionedTopic = topic + "-partition-0"; @@ -833,7 +876,7 @@ public void testExamineMessageMetadata() throws Exception { @Test public void testOffloadWithNullMessageId() { final String topicName = "topic-123"; - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true, null); try { persistentTopics.triggerOffload(testTenant, testNamespace, topicName, true, null); @@ -1107,7 +1150,7 @@ public void testDeleteTopic() throws Exception { final String topicName = "topic-1"; BrokerService brokerService = spy(pulsar.getBrokerService()); doReturn(brokerService).when(pulsar).getBrokerService(); - persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false); + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null); CompletableFuture deleteTopicFuture = new CompletableFuture<>(); deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException()); doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index 0c042f543188b..ff830e3edc8c6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -1051,7 +1051,7 @@ public void testTopicLoadingOnDisableNamespaceBundle() throws Exception { // try to create topic which should fail as bundle is disable CompletableFuture> futureResult = pulsar.getBrokerService() - .loadOrCreatePersistentTopic(topicName, true); + .loadOrCreatePersistentTopic(topicName, true, null); try { futureResult.get(); diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java index 1c072c8e93540..5564632edc018 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java @@ -344,7 +344,43 @@ List getListInBundle(String namespace, String bundleRange) * Number of partitions to create of the topic * @throws PulsarAdminException */ - void createPartitionedTopic(String topic, int numPartitions) throws PulsarAdminException; + default void createPartitionedTopic(String topic, int numPartitions) throws PulsarAdminException { + createPartitionedTopic(topic, numPartitions, null); + } + + /** + * Create a partitioned topic. + *

+ * Create a partitioned topic. It needs to be called before creating a producer for a partitioned topic. + *

+ * + * @param topic + * Topic name + * @param numPartitions + * Number of partitions to create of the topic + * @param properties + * topic properties + * @throws PulsarAdminException + */ + void createPartitionedTopic(String topic, int numPartitions, Map properties) + throws PulsarAdminException; + + /** + * Create a partitioned topic asynchronously. + *

+ * Create a partitioned topic asynchronously. It needs to be called before creating a producer for a partitioned + * topic. + *

+ * + * @param topic + * Topic name + * @param numPartitions + * Number of partitions to create of the topic + * @return a future that can be used to track when the partitioned topic is created + */ + default CompletableFuture createPartitionedTopicAsync(String topic, int numPartitions) { + return createPartitionedTopicAsync(topic, numPartitions, null); + } /** * Create a partitioned topic asynchronously. @@ -357,9 +393,25 @@ List getListInBundle(String namespace, String bundleRange) * Topic name * @param numPartitions * Number of partitions to create of the topic + * @param properties + * Topic properties * @return a future that can be used to track when the partitioned topic is created */ - CompletableFuture createPartitionedTopicAsync(String topic, int numPartitions); + CompletableFuture createPartitionedTopicAsync(String topic, int numPartitions, + Map properties); + + /** + * Create a non-partitioned topic. + *

+ * Create a non-partitioned topic. + *

+ * + * @param topic Topic name + * @throws PulsarAdminException + */ + default void createNonPartitionedTopic(String topic) throws PulsarAdminException { + createNonPartitionedTopic(topic, null); + } /** * Create a non-partitioned topic. @@ -368,16 +420,27 @@ List getListInBundle(String namespace, String bundleRange) *

* * @param topic Topic name + * @param properties Topic properties * @throws PulsarAdminException */ - void createNonPartitionedTopic(String topic) throws PulsarAdminException; + void createNonPartitionedTopic(String topic, Map properties) throws PulsarAdminException; + + /** + * Create a non-partitioned topic asynchronously. + * + * @param topic Topic name + */ + default CompletableFuture createNonPartitionedTopicAsync(String topic) { + return createNonPartitionedTopicAsync(topic, null); + } /** * Create a non-partitioned topic asynchronously. * * @param topic Topic name + * @param properties Topic properties */ - CompletableFuture createNonPartitionedTopicAsync(String topic); + CompletableFuture createNonPartitionedTopicAsync(String topic, Map properties); /** * Create missed partitions for partitioned topic. diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/partition/PartitionedTopicMetadata.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/partition/PartitionedTopicMetadata.java index 024ca0a32cef0..7d3fda3ef0b09 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/partition/PartitionedTopicMetadata.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/partition/PartitionedTopicMetadata.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.common.partition; +import java.util.Map; + /** * Metadata of a partitioned topic. */ @@ -26,12 +28,21 @@ public class PartitionedTopicMetadata { /* Number of partitions for the topic */ public int partitions; + /* Topic properties */ + public Map properties; + public PartitionedTopicMetadata() { - this.partitions = 0; + this(0); } public PartitionedTopicMetadata(int partitions) { this.partitions = partitions; + this.properties = null; + } + + public PartitionedTopicMetadata(int partitions, Map properties) { + this.partitions = partitions; + this.properties = properties; } /** diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java index 2901c926da425..7afbf9ae3dbe4 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java @@ -95,6 +95,7 @@ public class TopicsImpl extends BaseResource implements Topics { private final WebTarget adminTopics; private final WebTarget adminV2Topics; + private final WebTarget adminV3Topics; // CHECKSTYLE.OFF: MemberName private static final String BATCH_HEADER = "X-Pulsar-num-batch-message"; private static final String BATCH_SIZE_HEADER = "X-Pulsar-batch-size"; @@ -132,6 +133,7 @@ public TopicsImpl(WebTarget web, Authentication auth, long readTimeoutMs) { super(auth, readTimeoutMs); adminTopics = web.path("/admin"); adminV2Topics = web.path("/admin/v2"); + adminV3Topics = web.path("/admin/v3"); } @Override @@ -312,13 +314,14 @@ public CompletableFuture revokePermissionsAsync(String topic, String role) } @Override - public void createPartitionedTopic(String topic, int numPartitions) throws PulsarAdminException { - sync(() -> createPartitionedTopicAsync(topic, numPartitions)); + public void createPartitionedTopic(String topic, int numPartitions, Map metadata) + throws PulsarAdminException { + sync(() -> createPartitionedTopicAsync(topic, numPartitions, metadata)); } @Override - public void createNonPartitionedTopic(String topic) throws PulsarAdminException { - sync(() -> createNonPartitionedTopicAsync(topic)); + public void createNonPartitionedTopic(String topic, Map metadata) throws PulsarAdminException { + sync(() -> createNonPartitionedTopicAsync(topic, metadata)); } @Override @@ -327,24 +330,33 @@ public void createMissedPartitions(String topic) throws PulsarAdminException { } @Override - public CompletableFuture createNonPartitionedTopicAsync(String topic) { + public CompletableFuture createNonPartitionedTopicAsync(String topic, Map properties){ TopicName tn = validateTopic(topic); WebTarget path = topicPath(tn); - return asyncPutRequest(path, Entity.entity("", MediaType.APPLICATION_JSON)); + properties = properties == null ? new HashMap<>() : properties; + return asyncPutRequest(path, Entity.entity(properties, MediaType.APPLICATION_JSON)); } @Override - public CompletableFuture createPartitionedTopicAsync(String topic, int numPartitions) { - return createPartitionedTopicAsync(topic, numPartitions, false); + public CompletableFuture createPartitionedTopicAsync(String topic, int numPartitions, + Map properties) { + return createPartitionedTopicAsync(topic, numPartitions, false, properties); } public CompletableFuture createPartitionedTopicAsync( - String topic, int numPartitions, boolean createLocalTopicOnly) { + String topic, int numPartitions, boolean createLocalTopicOnly, Map properties) { checkArgument(numPartitions > 0, "Number of partitions should be more than 0"); TopicName tn = validateTopic(topic); - WebTarget path = topicPath(tn, "partitions") + WebTarget path = topicPath(tn, properties, "partitions") .queryParam("createLocalTopicOnly", Boolean.toString(createLocalTopicOnly)); - return asyncPutRequest(path, Entity.entity(numPartitions, MediaType.APPLICATION_JSON)); + Entity entity; + if (properties != null) { + PartitionedTopicMetadata metadata = new PartitionedTopicMetadata(numPartitions, properties); + entity = Entity.entity(metadata, MediaType.APPLICATION_JSON); + } else { + entity = Entity.entity(numPartitions, MediaType.APPLICATION_JSON); + } + return asyncPutRequest(path, entity); } @Override @@ -1239,6 +1251,22 @@ private WebTarget namespacePath(String domain, NamespaceName namespace, String.. return namespacePath; } + /** + * As we support topic metadata, user can add some properties when create topic. + * For compatibility, we have to define a new method, so when metadata is not null, v3 will be called. + * Details could be found here : https://github.com/apache/pulsar/pull/12818#discussion_r789340203 + * @param topic + * @param metadata + * @param parts + * @return + */ + private WebTarget topicPath(TopicName topic, Map metadata, String... parts) { + final WebTarget base = metadata != null ? adminV3Topics : (topic.isV2() ? adminV2Topics : adminTopics); + WebTarget topicPath = base.path(topic.getRestPath()); + topicPath = WebTargets.addParts(topicPath, parts); + return topicPath; + } + private WebTarget topicPath(TopicName topic, String... parts) { final WebTarget base = topic.isV2() ? adminV2Topics : adminTopics; WebTarget topicPath = base.path(topic.getRestPath()); diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java index 05636f59984d1..0120861fc6605 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java @@ -1396,13 +1396,13 @@ public void topics() throws Exception { verify(mockTopics).createSubscription("persistent://myprop/clust/ns1/ds1", "sub1", MessageId.earliest); cmdTopics.run(split("create-partitioned-topic persistent://myprop/clust/ns1/ds1 --partitions 32")); - verify(mockTopics).createPartitionedTopic("persistent://myprop/clust/ns1/ds1", 32); + verify(mockTopics).createPartitionedTopic("persistent://myprop/clust/ns1/ds1", 32, new HashMap<>()); cmdTopics.run(split("create-missed-partitions persistent://myprop/clust/ns1/ds1")); verify(mockTopics).createMissedPartitions("persistent://myprop/clust/ns1/ds1"); cmdTopics.run(split("create persistent://myprop/clust/ns1/ds1")); - verify(mockTopics).createNonPartitionedTopic("persistent://myprop/clust/ns1/ds1"); + verify(mockTopics).createNonPartitionedTopic("persistent://myprop/clust/ns1/ds1", new HashMap<>()); cmdTopics.run(split("list-partitioned-topics myprop/clust/ns1")); verify(mockTopics).getPartitionedTopicList("myprop/clust/ns1"); @@ -1835,7 +1835,7 @@ public void nonPersistentTopics() throws Exception { verify(mockTopics).getInternalStats("non-persistent://myprop/ns1/ds1", false); topics.run(split("create-partitioned-topic non-persistent://myprop/ns1/ds1 --partitions 32")); - verify(mockTopics).createPartitionedTopic("non-persistent://myprop/ns1/ds1", 32); + verify(mockTopics).createPartitionedTopic("non-persistent://myprop/ns1/ds1", 32, new HashMap<>()); topics.run(split("list myprop/ns1")); verify(mockTopics).getList("myprop/ns1", null); diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java index e5127f96eedb5..d2a7fce2b3e4d 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java @@ -499,10 +499,29 @@ private class CreatePartitionedCmd extends CliCommand { "--partitions" }, description = "Number of partitions for the topic", required = true) private int numPartitions; + @Parameter(names = {"--metadata", "-m"}, description = "key value pair properties(a=a,b=b,c=c)") + private java.util.List metadata; + @Override void run() throws Exception { String topic = validateTopicName(params); - getTopics().createPartitionedTopic(topic, numPartitions); + Map map = new HashMap<>(); + if (metadata != null) { + for (String property : metadata) { + if (!property.contains("=")) { + throw new ParameterException(String.format("Invalid key value pair '%s', " + + "valid format like 'a=a,b=b,c=c'.", property)); + } else { + String[] keyValue = property.split("="); + if (keyValue.length != 2) { + throw new ParameterException(String.format("Invalid key value pair '%s', " + + "valid format like 'a=a,b=b,c=c'.", property)); + } + map.put(keyValue[0], keyValue[1]); + } + } + } + getTopics().createPartitionedTopic(topic, numPartitions, map); } } @@ -527,10 +546,29 @@ private class CreateNonPartitionedCmd extends CliCommand { @Parameter(description = "persistent://tenant/namespace/topic", required = true) private java.util.List params; + @Parameter(names = {"--metadata", "-m"}, description = "key value pair properties(a=a,b=b,c=c)") + private java.util.List metadata; + @Override void run() throws Exception { String topic = validateTopicName(params); - getTopics().createNonPartitionedTopic(topic); + Map map = new HashMap<>(); + if (metadata != null) { + for (String property : metadata) { + if (!property.contains("=")) { + throw new ParameterException(String.format("Invalid key value pair '%s', " + + "valid format like 'a=a,b=b,c=c'.", property)); + } else { + String[] keyValue = property.split("="); + if (keyValue.length != 2) { + throw new ParameterException(String.format("Invalid key value pair '%s', " + + "valid format like 'a=a,b=b,c=c'.", property)); + } + map.put(keyValue[0], keyValue[1]); + } + } + } + getTopics().createNonPartitionedTopic(topic, map); } }