diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index f2327e28b93c8..f89951be8e981 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -24,7 +24,6 @@ import static org.apache.pulsar.common.policies.data.PoliciesUtil.getBundles; import com.google.common.collect.Lists; import com.google.common.collect.Sets; -import com.google.common.collect.Sets.SetView; import java.lang.reflect.Field; import java.net.URI; import java.net.URL; @@ -1961,34 +1960,6 @@ private void unsubscribe(NamespaceName nsName, String bundleRange, String subscr } } - /** - * It validates that peer-clusters can't coexist in replication-clusters. - * - * @param clusterName: given cluster whose peer-clusters can't be present into replication-cluster list - * @param replicationClusters: replication-cluster list - */ - private void validatePeerClusterConflict(String clusterName, Set replicationClusters) { - try { - ClusterData clusterData = clusterResources().getCluster(clusterName).orElseThrow( - () -> new RestException(Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName)); - Set peerClusters = clusterData.getPeerClusterNames(); - if (peerClusters != null && !peerClusters.isEmpty()) { - SetView conflictPeerClusters = Sets.intersection(peerClusters, replicationClusters); - if (!conflictPeerClusters.isEmpty()) { - log.warn("[{}] {}'s peer cluster can't be part of replication clusters {}", clientAppId(), - clusterName, conflictPeerClusters); - throw new RestException(Status.CONFLICT, - String.format("%s's peer-clusters %s can't be part of replication-clusters %s", clusterName, - conflictPeerClusters, replicationClusters)); - } - } - } catch (RestException re) { - throw re; - } catch (Exception e) { - log.warn("[{}] Failed to get cluster-data for {}", clientAppId(), clusterName, e); - } - } - protected BundlesData validateBundlesData(BundlesData initialBundles) { SortedSet partitions = new TreeSet(); for (String partition : initialBundles.getBoundaries()) { 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 ed1dc89530d51..fb466bf77cc0a 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 @@ -2816,6 +2816,60 @@ protected CompletableFuture internalSetBacklogQuota(BacklogQuota.BacklogQu }); } + protected CompletableFuture internalSetReplicationClusters(List clusterIds) { + validateTopicPolicyOperation(topicName, PolicyName.REPLICATION, PolicyOperation.WRITE); + validatePoliciesReadOnlyAccess(); + + Set replicationClusters = Sets.newHashSet(clusterIds); + if (replicationClusters.contains("global")) { + throw new RestException(Status.PRECONDITION_FAILED, + "Cannot specify global in the list of replication clusters"); + } + Set clusters = clusters(); + for (String clusterId : replicationClusters) { + if (!clusters.contains(clusterId)) { + throw new RestException(Status.FORBIDDEN, "Invalid cluster id: " + clusterId); + } + validatePeerClusterConflict(clusterId, replicationClusters); + validateClusterForTenant(namespaceName.getTenant(), clusterId); + } + + return getTopicPoliciesAsyncWithRetry(topicName).thenCompose(op -> { + TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new); + topicPolicies.setReplicationClusters(Lists.newArrayList(replicationClusters)); + return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies) + .thenRun(() -> { + log.info("[{}] Successfully set replication clusters for namespace={}, " + + "topic={}, clusters={}", + clientAppId(), + namespaceName, + topicName.getLocalName(), + topicPolicies.getReplicationClusters()); + }); + } + ); + } + + protected CompletableFuture internalRemoveReplicationClusters() { + validateTopicPolicyOperation(topicName, PolicyName.REPLICATION, PolicyOperation.WRITE); + validatePoliciesReadOnlyAccess(); + + return getTopicPoliciesAsyncWithRetry(topicName).thenCompose(op -> { + TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new); + topicPolicies.setReplicationClusters(null); + return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies) + .thenRun(() -> { + log.info("[{}] Successfully set replication clusters for namespace={}, " + + "topic={}, clusters={}", + clientAppId(), + namespaceName, + topicName.getLocalName(), + topicPolicies.getReplicationClusters()); + }); + } + ); + } + protected CompletableFuture internalGetDeduplication(boolean applied) { return getTopicPoliciesAsyncWithRetry(topicName) .thenApply(op -> op.map(TopicPolicies::getDeduplicationEnabled) 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 09b694cc34e02..1318057839955 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 @@ -1677,6 +1677,87 @@ public void removeBacklogQuota(@Suspended final AsyncResponse asyncResponse, }); } + @GET + @Path("/{tenant}/{namespace}/{topic}/replication") + @ApiOperation(value = "Get the replication clusters for a topic") + @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 405, message = + "Topic level policy is disabled, enable the topic level policy and retry")}) + public void getReplicationClusters(@Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, + @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @QueryParam("applied") boolean applied, + @ApiParam(value = "Is authentication required to perform this operation") + @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) { + validateTopicName(tenant, namespace, encodedTopic); + preValidation(authoritative) + .thenCompose(__ -> getTopicPoliciesAsyncWithRetry(topicName)) + .thenAccept(op -> { + asyncResponse.resume(op.map(TopicPolicies::getReplicationClustersSet).orElseGet(() -> { + if (applied) { + return getNamespacePolicies(namespaceName).replication_clusters; + } + return null; + })); + }) + .exceptionally(ex -> { + handleTopicPolicyException("getReplicationClusters", ex, asyncResponse); + return null; + }); + } + + @POST + @Path("/{tenant}/{namespace}/{topic}/replication") + @ApiOperation(value = "Set the replication clusters for a topic.") + @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 409, message = "Concurrent modification"), + @ApiResponse(code = 405, + message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 412, message = "Topic is not global or invalid cluster ids")}) + public void setReplicationClusters( + @Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @ApiParam(value = "Is authentication required to perform this operation") + @QueryParam("authoritative") @DefaultValue("false") boolean authoritative, + @ApiParam(value = "List of replication clusters", required = true) List clusterIds) { + validateTopicName(tenant, namespace, encodedTopic); + preValidation(authoritative) + .thenCompose(__ -> internalSetReplicationClusters(clusterIds)) + .thenRun(() -> asyncResponse.resume(Response.noContent().build())) + .exceptionally(ex -> { + handleTopicPolicyException("setReplicationClusters", ex, asyncResponse); + return null; + }); + } + + @DELETE + @Path("/{tenant}/{namespace}/{topic}/replication") + @ApiOperation(value = "Remove the replication clusters from a topic.") + @ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 405, + message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 409, message = "Concurrent modification")}) + public void removeReplicationClusters(@Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @QueryParam("backlogQuotaType") BacklogQuotaType backlogQuotaType, + @ApiParam(value = "Is authentication required to perform this operation") + @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) { + validateTopicName(tenant, namespace, encodedTopic); + preValidation(authoritative) + .thenCompose(__ -> internalRemoveReplicationClusters()) + .thenRun(() -> asyncResponse.resume(Response.noContent().build())) + .exceptionally(ex -> { + handleTopicPolicyException("removeReplicationClusters", ex, asyncResponse); + return null; + }); + } + @GET @Path("/{tenant}/{namespace}/{topic}/messageTTL") @ApiOperation(value = "Get message TTL in seconds for a topic") diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index b1ee3e3a1a1fd..71b94715089bf 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1341,33 +1341,15 @@ public CompletableFuture checkReplication() { log.debug("[{}] Checking replication status", name); } - CompletableFuture policiesFuture = brokerService.pulsar().getPulsarResources() - .getNamespaceResources() - .getPoliciesAsync(TopicName.get(topic).getNamespaceObject()) - .thenCompose(optPolicies -> { - if (!optPolicies.isPresent()) { - return FutureUtil.failedFuture( - new ServerMetadataException( - new MetadataStoreException.NotFoundException())); - } - - return CompletableFuture.completedFuture(optPolicies.get()); - }); + CompletableFuture> replicationClustersFuture = getReplicationClusters(name); CompletableFuture ttlFuture = getMessageTTL(); - return CompletableFuture.allOf(policiesFuture, ttlFuture) + return CompletableFuture.allOf(replicationClustersFuture, ttlFuture) .thenCompose(__ -> { - Policies policies = policiesFuture.join(); + List configuredClusters = replicationClustersFuture.join(); int newMessageTTLinSeconds = ttlFuture.join(); - Set configuredClusters; - if (policies.replication_clusters != null) { - configuredClusters = Sets.newTreeSet(policies.replication_clusters); - } else { - configuredClusters = Collections.emptySet(); - } - String localCluster = brokerService.pulsar().getConfiguration().getClusterName(); // if local cluster is removed from global namespace cluster-list : then delete topic forcefully @@ -1408,6 +1390,32 @@ public CompletableFuture checkReplication() { }); } + @VisibleForTesting + public CompletableFuture> getReplicationClusters(TopicName topicName) { + return brokerService.pulsar() + .getTopicPoliciesService() + .getTopicPoliciesAsyncWithRetry(topicName, null, brokerService.pulsar().getExecutor()) + .thenCompose(topicPolicies -> { + if (!topicPolicies.isPresent() || topicPolicies.get().getReplicationClusters() == null) { + return brokerService.pulsar().getPulsarResources() + .getNamespaceResources() + .getPoliciesAsync(TopicName.get(topic).getNamespaceObject()) + .thenCompose(optPolicies -> { + if (!optPolicies.isPresent()) { + return FutureUtil.failedFuture( + new ServerMetadataException( + new MetadataStoreException.NotFoundException())); + } + + return CompletableFuture.completedFuture( + Lists.newArrayList(optPolicies.get().replication_clusters)); + }); + } else { + return CompletableFuture.completedFuture(topicPolicies.get().getReplicationClusters()); + } + }); + } + @Override public void checkMessageExpiry() { getMessageTTL().thenAccept(messageTtlInSeconds -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index d4955de8ba999..e3ef2b3852838 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -25,6 +25,7 @@ import com.google.common.collect.BoundType; import com.google.common.collect.Lists; import com.google.common.collect.Range; +import com.google.common.collect.Sets; import java.net.MalformedURLException; import java.net.URI; import java.net.URL; @@ -306,6 +307,34 @@ protected static void validateAdminAccessForTenant(PulsarService pulsar, String } } + /** + * It validates that peer-clusters can't coexist in replication-clusters. + * + * @clusterName: given cluster whose peer-clusters can't be present into replication-cluster list + * @replicationClusters: replication-cluster list + */ + protected void validatePeerClusterConflict(String clusterName, Set replicationClusters) { + try { + ClusterData clusterData = clusterResources().getCluster(clusterName).orElseThrow( + () -> new RestException(Status.PRECONDITION_FAILED, "Invalid replication cluster " + clusterName)); + Set peerClusters = clusterData.getPeerClusterNames(); + if (peerClusters != null && !peerClusters.isEmpty()) { + Sets.SetView conflictPeerClusters = Sets.intersection(peerClusters, replicationClusters); + if (!conflictPeerClusters.isEmpty()) { + log.warn("[{}] {}'s peer cluster can't be part of replication clusters {}", clientAppId(), + clusterName, conflictPeerClusters); + throw new RestException(Status.CONFLICT, + String.format("%s's peer-clusters %s can't be part of replication-clusters %s", clusterName, + conflictPeerClusters, replicationClusters)); + } + } + } catch (RestException re) { + throw re; + } catch (Exception e) { + log.warn("[{}] Failed to get cluster-data for {}", clientAppId(), clusterName, e); + } + } + protected void validateClusterForTenant(String tenant, String cluster) { TenantInfo tenantInfo; try { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java index 0383333cb6002..535c4a425bca0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java @@ -24,6 +24,7 @@ import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; +import com.google.api.client.util.Lists; import com.google.common.collect.Maps; import com.google.common.collect.Sets; import java.lang.reflect.Field; @@ -2643,6 +2644,25 @@ public void testDoNotCreateSystemTopicForHeartbeatNamespace() { }); } + @Test(timeOut = 30000) + public void testReplicatorClusterApi() throws Exception { + final String topic = "persistent://" + myNamespace + "/test-" + UUID.randomUUID(); + // init cache + pulsarClient.newProducer().topic(topic).create().close(); + + assertNull(admin.topics().getReplicationClusters(topic, false)); + + List clusters = Lists.newArrayList(); + clusters.add("test"); + admin.topics().setReplicationClusters(topic, clusters); + Awaitility.await().untilAsserted(() + -> assertEquals(admin.topics().getReplicationClusters(topic, false), clusters)); + + admin.topics().removeReplicationClusters(topic); + Awaitility.await().untilAsserted(() + -> assertNull(admin.topics().getReplicationClusters(topic, false))); + } + @Test public void testLoopCreateAndDeleteTopicPolicies() throws Exception { final String topic = testTopic + UUID.randomUUID(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index 8603404a86a31..0b816e5a4748c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -46,9 +46,11 @@ import java.net.URL; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; @@ -117,6 +119,7 @@ import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.stats.SubscriptionStatsImpl; import org.apache.pulsar.common.protocol.ByteBufPair; import org.apache.pulsar.common.protocol.schema.SchemaVersion; @@ -2251,4 +2254,51 @@ private ByteBuf getMessageWithMetadata(byte[] data) { messageData.writeTo(headers); return ByteBufPair.coalesce(ByteBufPair.get(headers, payload)); } + + @Test + public void testGetReplicationClusters() throws Exception { + PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); + TopicName name = TopicName.get(successTopicName); + CompletableFuture> replicationClusters = topic.getReplicationClusters(name); + try { + replicationClusters.get(); + fail("Should have failed"); + } catch (ExecutionException ex) { + assertTrue(ex.getCause() instanceof BrokerServiceException.ServerMetadataException); + } + + PulsarResources pulsarResources = spy(new PulsarResources(store, store)); + NamespaceResources nsr = spy(new NamespaceResources(store, store, 30)); + doReturn(nsr).when(pulsarResources).getNamespaceResources(); + doReturn(pulsarResources).when(pulsar).getPulsarResources(); + CompletableFuture> policiesFuture = new CompletableFuture<>(); + Policies policies = new Policies(); + Set namespaceClusters = new HashSet<>(); + namespaceClusters.add("namespace-cluster"); + policies.replication_clusters = namespaceClusters; + Optional optionalPolicies = Optional.of(policies); + policiesFuture.complete(optionalPolicies); + doReturn(policiesFuture).when(nsr).getPoliciesAsync(any()); + + topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); + replicationClusters = topic.getReplicationClusters(name); + + assertEquals(replicationClusters.get(), namespaceClusters); + + TopicPoliciesService topicPoliciesService = mock(TopicPoliciesService.class); + doReturn(topicPoliciesService).when(pulsar).getTopicPoliciesService(); + CompletableFuture> topicPoliciesFuture = new CompletableFuture<>(); + TopicPolicies topicPolicies = new TopicPolicies(); + List topicClusters = new ArrayList<>(); + topicClusters.add("topic-cluster"); + topicPolicies.setReplicationClusters(topicClusters); + Optional optionalTopicPolicies = Optional.of(topicPolicies); + topicPoliciesFuture.complete(optionalTopicPolicies); + when(topicPoliciesService.getTopicPoliciesAsyncWithRetry(any(), any(), any())).thenReturn(topicPoliciesFuture); + + topic = new PersistentTopic(successTopicName, ledgerMock, brokerService); + replicationClusters = topic.getReplicationClusters(name); + + assertEquals(replicationClusters.get(), topicClusters); + } } 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 daab4094b1258..03bca5cfcb2ef 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 @@ -3580,4 +3580,62 @@ CompletableFuture> examineMessageAsync(String topic, String init * @return */ CompletableFuture setReplicatedSubscriptionStatusAsync(String topic, String subName, boolean enabled); + + /** + * Get the replication clusters for a topic. + * + * @param topic + * @param applied + * @return + * @throws PulsarAdminException + */ + Set getReplicationClusters(String topic, boolean applied) throws PulsarAdminException; + + /** + * Get the replication clusters for a topic asynchronously. + * + * @param topic + * @param applied + * @return + * @throws PulsarAdminException + */ + CompletableFuture> getReplicationClustersAsync(String topic, boolean applied); + + /** + * Set the replication clusters for the topic. + * + * @param topic + * @param clusterIds + * @return + * @throws PulsarAdminException + */ + void setReplicationClusters(String topic, List clusterIds) throws PulsarAdminException; + + /** + * Set the replication clusters for the topic asynchronously. + * + * @param topic + * @param clusterIds + * @return + * @throws PulsarAdminException + */ + CompletableFuture setReplicationClustersAsync(String topic, List clusterIds); + + /** + * Remove the replication clusters for the topic. + * + * @param topic + * @return + * @throws PulsarAdminException + */ + void removeReplicationClusters(String topic) throws PulsarAdminException; + + /** + * Remove the replication clusters for the topic asynchronously. + * + * @param topic + * @return + * @throws PulsarAdminException + */ + CompletableFuture removeReplicationClustersAsync(String topic); } 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 caf32e4aacf8a..7cd153f260b74 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 @@ -3832,5 +3832,82 @@ public CompletableFuture setReplicatedSubscriptionStatusAsync(String topic return asyncPostRequest(path, Entity.entity(enabled, MediaType.APPLICATION_JSON)); } + @Override + public Set getReplicationClusters(String topic, boolean applied) throws PulsarAdminException { + try { + return getReplicationClustersAsync(topic, applied).get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture> getReplicationClustersAsync(String topic, boolean applied) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "replication"); + path = path.queryParam("applied", applied); + final CompletableFuture> future = new CompletableFuture<>(); + asyncGetRequest(path, + new InvocationCallback>() { + @Override + public void completed(Set clusterIds) { + future.complete(clusterIds); + } + + @Override + public void failed(Throwable throwable) { + future.completeExceptionally(getApiException(throwable.getCause())); + } + }); + return future; + } + + @Override + public void setReplicationClusters(String topic, List clusterIds) throws PulsarAdminException { + try { + setReplicationClustersAsync(topic, clusterIds).get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture setReplicationClustersAsync(String topic, List clusterIds) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "replication"); + return asyncPostRequest(path, Entity.entity(clusterIds, MediaType.APPLICATION_JSON)); + } + + @Override + public void removeReplicationClusters(String topic) throws PulsarAdminException { + try { + removeReplicationClustersAsync(topic).get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture removeReplicationClustersAsync(String topic) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "replication"); + return asyncDeleteRequest(path); + } + private static final Logger log = LoggerFactory.getLogger(TopicsImpl.class); } 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 b311417f22b10..ba6153ae54f5d 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 @@ -1295,6 +1295,15 @@ public boolean matches(Long timestamp) { cmdTopics.run(split("get-max-consumers persistent://myprop/clust/ns1/ds1 -ap")); verify(mockTopics).getMaxConsumers("persistent://myprop/clust/ns1/ds1", true); + + cmdTopics.run(split("get-replication-clusters persistent://myprop/clust/ns1/ds1 --applied")); + verify(mockTopics).getReplicationClusters("persistent://myprop/clust/ns1/ds1", true); + + cmdTopics.run(split("set-replication-clusters persistent://myprop/clust/ns1/ds1 -c test")); + verify(mockTopics).setReplicationClusters("persistent://myprop/clust/ns1/ds1", Lists.newArrayList("test")); + + cmdTopics.run(split("remove-replication-clusters persistent://myprop/clust/ns1/ds1")); + verify(mockTopics).removeReplicationClusters("persistent://myprop/clust/ns1/ds1"); } private static LedgerInfo newLedger(long id, long entries, long size) { 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 9fa37816b34c8..7ba4f7379e72d 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 @@ -227,6 +227,10 @@ public CmdTopics(Supplier admin) { jcommander.addCommand("set-replicated-subscription-status", new SetReplicatedSubscriptionStatus()); jcommander.addCommand("get-backlog-size", new GetBacklogSizeByMessageId()); + jcommander.addCommand("get-replication-clusters", new GetReplicationClusters()); + jcommander.addCommand("set-replication-clusters", new SetReplicationClusters()); + jcommander.addCommand("remove-replication-clusters", new RemoveReplicationClusters()); + initDeprecatedCommands(); } @@ -1252,6 +1256,50 @@ void run() throws PulsarAdminException { } } + @Parameters(commandDescription = "Get the replication clusters for a topic") + private class GetReplicationClusters extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = { "-ap", "--applied" }, description = "Get the applied policy of the topic") + private boolean applied = false; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + print(getTopics().getReplicationClusters(persistentTopic, applied)); + } + } + + @Parameters(commandDescription = "Set the replication clusters for a topic") + private class SetReplicationClusters extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = { "--clusters", + "-c" }, description = "Replication Cluster Ids list (comma separated values)", required = true) + private String clusterIds; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + List clusters = Lists.newArrayList(clusterIds.split(",")); + getTopics().setReplicationClusters(persistentTopic, clusters); + } + } + + @Parameters(commandDescription = "Remove the replication clusters for a topic") + private class RemoveReplicationClusters extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + getTopics().removeReplicationClusters(persistentTopic); + } + } + @Parameters(commandDescription = "Get the delayed delivery policy for a topic") private class GetDelayedDelivery extends CliCommand { @Parameter(description = "tenant/namespace/topic", required = true) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java index 1dff1c4a906ee..8b5d39129eb4b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java @@ -18,16 +18,17 @@ */ package org.apache.pulsar.common.policies.data; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; -import lombok.Getter; import lombok.NoArgsConstructor; -import lombok.Setter; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl; import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl; @@ -40,15 +41,13 @@ @Builder @NoArgsConstructor @AllArgsConstructor -@Getter -@Setter public class TopicPolicies { @Builder.Default private Map backLogQuotaMap = new HashMap<>(); @Builder.Default private List subscriptionTypesEnabled = new ArrayList<>(); - + private List replicationClusters; private PersistencePolicies persistence; private RetentionPolicies retentionPolicies; private Boolean deduplicationEnabled; @@ -163,4 +162,8 @@ public boolean isPublishRateSet() { public boolean isSubscribeRateSet() { return subscribeRate != null; } + + public Set getReplicationClustersSet() { + return replicationClusters != null ? Sets.newTreeSet(this.replicationClusters) : null; + } }