You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2020/03/24 02:04:10 UTC
[pulsar] branch master updated: [PulsarAdmin] Topics to async
(#6593)
This is an automated email from the ASF dual-hosted git repository.
zhaijia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1537de2 [PulsarAdmin] Topics to async (#6593)
1537de2 is described below
commit 1537de23d71a3fcf965741f8649e752dacd30d49
Author: Yijie Shen <he...@gmail.com>
AuthorDate: Tue Mar 24 10:03:57 2020 +0800
[PulsarAdmin] Topics to async (#6593)
---
.../org/apache/pulsar/client/admin/Topics.java | 250 ++++++++++---
.../pulsar/client/admin/internal/TopicsImpl.java | 386 ++++++++++++++++-----
2 files changed, 513 insertions(+), 123 deletions(-)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
index 1b2f3fc..17a9e44 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
@@ -65,6 +65,22 @@ public interface Topics {
List<String> getList(String namespace) throws PulsarAdminException;
/**
+ * Get the list of topics under a namespace asynchronously.
+ * <p>
+ * Response example:
+ *
+ * <pre>
+ * <code>["topic://my-tenant/my-namespace/topic-1",
+ * "topic://my-tenant/my-namespace/topic-2"]</code>
+ * </pre>
+ *
+ * @param namespace
+ * Namespace name
+ * @return a list of topics
+ */
+ CompletableFuture<List<String>> getListAsync(String namespace);
+
+ /**
* Get the list of partitioned topics under a namespace.
* <p>
* Response example:
@@ -88,6 +104,22 @@ public interface Topics {
List<String> getPartitionedTopicList(String namespace) throws PulsarAdminException;
/**
+ * Get the list of partitioned topics under a namespace asynchronously.
+ * <p>
+ * Response example:
+ *
+ * <pre>
+ * <code>["persistent://my-tenant/my-namespace/topic-1",
+ * "persistent://my-tenant/my-namespace/topic-2"]</code>
+ * </pre>
+ *
+ * @param namespace
+ * Namespace name
+ * @return a list of partitioned topics
+ */
+ CompletableFuture<List<String>> getPartitionedTopicListAsync(String namespace);
+
+ /**
* Get list of topics exist into given bundle
*
* @param namespace
@@ -136,6 +168,27 @@ public interface Topics {
Map<String, Set<AuthAction>> getPermissions(String topic) throws PulsarAdminException;
/**
+ * Get permissions on a topic asynchronously.
+ * <p>
+ * Retrieve the effective permissions for a topic. These permissions are defined by the permissions set at the
+ * namespace level combined (union) with any eventual specific permission set on the topic.
+ * <p>
+ * Response Example:
+ *
+ * <pre>
+ * <code>{
+ * "role-1" : [ "produce" ],
+ * "role-2" : [ "consume" ]
+ * }</code>
+ * </pre>
+ *
+ * @param topic
+ * Topic url
+ * @return a map of topics an their permissions set
+ */
+ CompletableFuture<Map<String, Set<AuthAction>>> getPermissionsAsync(String topic);
+
+ /**
* Grant permission on a topic.
* <p>
* Grant a new permission to a client role on a single topic.
@@ -165,6 +218,26 @@ public interface Topics {
void grantPermission(String topic, String role, Set<AuthAction> actions) throws PulsarAdminException;
/**
+ * Grant permission on a topic asynchronously.
+ * <p>
+ * Grant a new permission to a client role on a single topic.
+ * <p>
+ * Request parameter example:
+ *
+ * <pre>
+ * <code>["produce", "consume"]</code>
+ * </pre>
+ *
+ * @param topic
+ * Topic url
+ * @param role
+ * Client role to which grant permission
+ * @param actions
+ * Auth actions (produce and consume)
+ */
+ CompletableFuture<Void> grantPermissionAsync(String topic, String role, Set<AuthAction> actions);
+
+ /**
* Revoke permissions on a topic.
* <p>
* Revoke permissions to a client role on a single topic. If the permission was not set at the topic level, but
@@ -186,6 +259,19 @@ public interface Topics {
void revokePermissions(String topic, String role) throws PulsarAdminException;
/**
+ * Revoke permissions on a topic asynchronously.
+ * <p>
+ * Revoke permissions to a client role on a single topic. If the permission was not set at the topic level, but
+ * rather at the namespace level, this operation will return an error (HTTP status code 412).
+ *
+ * @param topic
+ * Topic url
+ * @param role
+ * Client role to which remove permission
+ */
+ CompletableFuture<Void> revokePermissionsAsync(String topic, String role);
+
+ /**
* Create a partitioned topic.
* <p>
* Create a partitioned topic. It needs to be called before creating a producer for a partitioned topic.
@@ -200,6 +286,21 @@ public interface Topics {
void createPartitionedTopic(String topic, int numPartitions) throws PulsarAdminException;
/**
+ * Create a partitioned topic asynchronously.
+ * <p>
+ * Create a partitioned topic asynchronously. It needs to be called before creating a producer for a partitioned
+ * topic.
+ * <p>
+ *
+ * @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
+ */
+ CompletableFuture<Void> createPartitionedTopicAsync(String topic, int numPartitions);
+
+ /**
* Create a non-partitioned topic.
*
* <p>
@@ -212,6 +313,13 @@ public interface Topics {
void createNonPartitionedTopic(String topic) throws PulsarAdminException;
/**
+ * Create a non-partitioned topic asynchronously.
+ *
+ * @param topic Topic name
+ */
+ CompletableFuture<Void> createNonPartitionedTopicAsync(String topic);
+
+ /**
* Create missed partitions for partitioned topic.
* <p>
* When disable topic auto creation, use this method to try create missed partitions while
@@ -222,28 +330,6 @@ public interface Topics {
void createMissedPartitions(String topic) throws PulsarAdminException;
/**
- * Create a partitioned topic asynchronously.
- * <p>
- * Create a partitioned topic asynchronously. It needs to be called before creating a producer for a partitioned
- * topic.
- * <p>
- *
- * @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
- */
- CompletableFuture<Void> createPartitionedTopicAsync(String topic, int numPartitions);
-
- /**
- * Create a non-partitioned topic asynchronously.
- *
- * @param topic Topic name
- */
- CompletableFuture<Void> createNonPartitionedTopicAsync(String topic);
-
- /**
* Create missed partitions for partitioned topic asynchronously.
* <p>
* When disable topic auto creation, use this method to try create missed partitions while
@@ -284,7 +370,25 @@ public interface Topics {
* @return a future that can be used to track when the partitioned topic is updated
*/
CompletableFuture<Void> updatePartitionedTopicAsync(String topic, int numPartitions);
-
+
+ /**
+ * Update number of partitions of a non-global partitioned topic.
+ * <p>
+ * It requires partitioned-topic to be already exist and number of new partitions must be greater than existing
+ * number of partitions. Decrementing number of partitions requires deletion of topic which is not supported.
+ * <p>
+ *
+ * @param topic
+ * Topic name
+ * @param numPartitions
+ * Number of new partitions of already exist partitioned-topic
+ * @param updateLocalTopicOnly
+ * Used by broker for global topic with multiple replicated clusters
+ *
+ * @return a future that can be used to track when the partitioned topic is updated
+ */
+ void updatePartitionedTopic(String topic, int numPartitions, boolean updateLocalTopicOnly) throws PulsarAdminException;
+
/**
* Update number of partitions of a non-global partitioned topic asynchronously.
* <p>
@@ -342,6 +446,21 @@ public interface Topics {
* @throws PulsarAdminException
*/
void deletePartitionedTopic(String topic, boolean force) throws PulsarAdminException;
+
+ /**
+ * Delete a partitioned topic asynchronously.
+ * <p>
+ * It will also delete all the partitions of the topic if it exists.
+ * <p>
+ *
+ * @param topic
+ * Topic name
+ * @param force
+ * Delete topic forcefully
+ *
+ * @return a future that can be used to track when the partitioned topic is deleted
+ */
+ CompletableFuture<Void> deletePartitionedTopicAsync(String topic, boolean force);
/**
* Delete a partitioned topic.
@@ -364,12 +483,8 @@ public interface Topics {
*
* @param topic
* Topic name
- * @param force
- * Delete topic forcefully
- *
- * @return a future that can be used to track when the partitioned topic is deleted
*/
- CompletableFuture<Void> deletePartitionedTopicAsync(String topic, boolean force);
+ CompletableFuture<Void> deletePartitionedTopicAsync(String topic);
/**
* Delete a topic.
@@ -393,6 +508,22 @@ public interface Topics {
*/
void delete(String topic, boolean force) throws PulsarAdminException;
+ /**
+ * Delete a topic asynchronously.
+ * <p>
+ * Delete a topic asynchronously. The topic cannot be deleted if force flag is disable and there's any active
+ * subscription or producer connected to the it. Force flag deletes topic forcefully by closing all active producers
+ * and consumers.
+ * <p>
+ *
+ * @param topic
+ * topic name
+ * @param force
+ * Delete topic forcefully
+ *
+ * @return a future that can be used to track when the topic is deleted
+ */
+ CompletableFuture<Void> deleteAsync(String topic, boolean force);
/**
* Delete a topic.
@@ -413,23 +544,17 @@ public interface Topics {
* Unexpected error
*/
void delete(String topic) throws PulsarAdminException;
-
+
/**
* Delete a topic asynchronously.
* <p>
- * Delete a topic asynchronously. The topic cannot be deleted if force flag is disable and there's any active
- * subscription or producer connected to the it. Force flag deletes topic forcefully by closing all active producers
- * and consumers.
+ * Delete a topic. The topic cannot be deleted if there's any active subscription or producer connected to the it.
* <p>
*
* @param topic
- * topic name
- * @param force
- * Delete topic forcefully
- *
- * @return a future that can be used to track when the topic is deleted
+ * Topic name
*/
- CompletableFuture<Void> deleteAsync(String topic, boolean force);
+ CompletableFuture<Void> deleteAsync(String topic);
/**
* Unload a topic.
@@ -461,7 +586,23 @@ public interface Topics {
/**
* Terminate the topic and prevent any more messages being published on it.
* <p>
- * This
+ *
+ * @param topic
+ * topic name
+ * @return the message id of the last message that was published in the topic
+ *
+ * @throws NotAuthorizedException
+ * Don't have admin permission
+ * @throws NotFoundException
+ * topic does not exist
+ * @throws PulsarAdminException
+ * Unexpected error
+ */
+ MessageId terminateTopic(String topic) throws PulsarAdminException;
+
+ /**
+ * Terminate the topic and prevent any more messages being published on it.
+ * <p>
*
* @param topic
* topic name
@@ -1085,6 +1226,13 @@ public interface Topics {
LongRunningProcessStatus compactionStatus(String topic) throws PulsarAdminException;
/**
+ * Check the status of an ongoing compaction for a topic asynchronously.
+ *
+ * @param topic The topic whose compaction status we wish to check
+ */
+ CompletableFuture<LongRunningProcessStatus> compactionStatusAsync(String topic);
+
+ /**
* Trigger offloading messages in topic to longterm storage.
*
* @param topic the topic to offload
@@ -1093,6 +1241,14 @@ public interface Topics {
void triggerOffload(String topic, MessageId messageId) throws PulsarAdminException;
/**
+ * Trigger offloading messages in topic to longterm storage asynchronously.
+ *
+ * @param topic the topic to offload
+ * @param messageId ID of maximum message which should be offloaded
+ */
+ CompletableFuture<Void> triggerOffloadAsync(String topic, MessageId messageId);
+
+ /**
* Check the status of an ongoing offloading operation for a topic.
*
* @param topic the topic being offloaded
@@ -1101,6 +1257,14 @@ public interface Topics {
OffloadProcessStatus offloadStatus(String topic) throws PulsarAdminException;
/**
+ * Check the status of an ongoing offloading operation for a topic asynchronously.
+ *
+ * @param topic the topic being offloaded
+ * @return the status of the offload operation
+ */
+ CompletableFuture<OffloadProcessStatus> offloadStatusAsync(String topic);
+
+ /**
* Get the last commit message Id of a topic
*
* @param topic the topic name
@@ -1108,4 +1272,12 @@ public interface Topics {
* @throws PulsarAdminException
*/
MessageId getLastMessageId(String topic) throws PulsarAdminException;
+
+ /**
+ * Get the last commit message Id of a topic asynchronously.
+ *
+ * @param topic the topic name
+ * @return
+ */
+ CompletableFuture<MessageId> getLastMessageIdAsync(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 aebdbb2..163c9bc 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
@@ -93,39 +93,101 @@ public class TopicsImpl extends BaseResource implements Topics {
@Override
public List<String> getList(String namespace) throws PulsarAdminException {
try {
- NamespaceName ns = NamespaceName.get(namespace);
- WebTarget persistentPath = namespacePath("persistent", ns);
- WebTarget nonPersistentPath = namespacePath("non-persistent", ns);
+ return getListAsync(namespace).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);
+ }
+ }
- List<String> persistentTopics = request(persistentPath).get(new GenericType<List<String>>() {
- });
+ @Override
+ public CompletableFuture<List<String>> getListAsync(String namespace) {
+ NamespaceName ns = NamespaceName.get(namespace);
+ WebTarget persistentPath = namespacePath("persistent", ns);
+ WebTarget nonPersistentPath = namespacePath("non-persistent", ns);
+ final CompletableFuture<List<String>> persistentList = new CompletableFuture<>();
+ final CompletableFuture<List<String>> nonPersistentList = new CompletableFuture<>();
+ asyncGetRequest(persistentPath,
+ new InvocationCallback<List<String>>() {
+ @Override
+ public void completed(List<String> topics) {
+ persistentList.complete(topics);
+ }
- List<String> nonPersistentTopics = request(nonPersistentPath).get(new GenericType<List<String>>() {
- });
- return new ArrayList<>(
- Stream.concat(persistentTopics.stream(), nonPersistentTopics.stream()).collect(Collectors.toSet()));
- } catch (Exception e) {
- throw getApiException(e);
- }
+ @Override
+ public void failed(Throwable throwable) {
+ persistentList.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ asyncGetRequest(nonPersistentPath,
+ new InvocationCallback<List<String>>() {
+ @Override
+ public void completed(List<String> a) {
+ nonPersistentList.complete(a);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ nonPersistentList.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+
+ return persistentList.thenCombine(nonPersistentList,
+ (l1, l2) -> new ArrayList<>(Stream.concat(l1.stream(), l2.stream()).collect(Collectors.toSet())));
}
@Override
public List<String> getPartitionedTopicList(String namespace) throws PulsarAdminException {
try {
- NamespaceName ns = NamespaceName.get(namespace);
+ return getPartitionedTopicListAsync(namespace).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);
+ }
+ }
- WebTarget persistentPath = namespacePath("persistent", ns, "partitioned");
- WebTarget nonPersistentPath = namespacePath("non-persistent", ns, "partitioned");
- List<String> persistentTopics = request(persistentPath).get(new GenericType<List<String>>() {
- });
+ @Override
+ public CompletableFuture<List<String>> getPartitionedTopicListAsync(String namespace) {
+ NamespaceName ns = NamespaceName.get(namespace);
+ WebTarget persistentPath = namespacePath("persistent", ns, "partitioned");
+ WebTarget nonPersistentPath = namespacePath("non-persistent", ns, "partitioned");
+ final CompletableFuture<List<String>> persistentList = new CompletableFuture<>();
+ final CompletableFuture<List<String>> nonPersistentList = new CompletableFuture<>();
+ asyncGetRequest(persistentPath,
+ new InvocationCallback<List<String>>() {
+ @Override
+ public void completed(List<String> topics) {
+ persistentList.complete(topics);
+ }
- List<String> nonPersistentTopics = request(nonPersistentPath).get(new GenericType<List<String>>() {
- });
- return new ArrayList<>(
- Stream.concat(persistentTopics.stream(), nonPersistentTopics.stream()).collect(Collectors.toSet()));
- } catch (Exception e) {
- throw getApiException(e);
- }
+ @Override
+ public void failed(Throwable throwable) {
+ persistentList.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ asyncGetRequest(nonPersistentPath,
+ new InvocationCallback<List<String>>() {
+ @Override
+ public void completed(List<String> topics) {
+ nonPersistentList.complete(topics);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ nonPersistentList.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+
+ return persistentList.thenCombine(nonPersistentList,
+ (l1, l2) -> new ArrayList<>(Stream.concat(l1.stream(), l2.stream()).collect(Collectors.toSet())));
}
@@ -167,37 +229,80 @@ public class TopicsImpl extends BaseResource implements Topics {
@Override
public Map<String, Set<AuthAction>> getPermissions(String topic) throws PulsarAdminException {
try {
- TopicName tn = TopicName.get(topic);
- WebTarget path = topicPath(tn, "permissions");
- return request(path).get(new GenericType<Map<String, Set<AuthAction>>>() {});
- } catch (Exception e) {
- throw getApiException(e);
+ return getPermissionsAsync(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<Map<String, Set<AuthAction>>> getPermissionsAsync(String topic) {
+ TopicName tn = TopicName.get(topic);
+ WebTarget path = topicPath(tn, "permissions");
+ final CompletableFuture<Map<String, Set<AuthAction>>> future = new CompletableFuture<>();
+ asyncGetRequest(path,
+ new InvocationCallback<Map<String, Set<AuthAction>>>() {
+ @Override
+ public void completed(Map<String, Set<AuthAction>> permissions) {
+ future.complete(permissions);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ return future;
+ }
+
+ @Override
public void grantPermission(String topic, String role, Set<AuthAction> actions) throws PulsarAdminException {
try {
- TopicName tn = TopicName.get(topic);
- WebTarget path = topicPath(tn, "permissions", role);
- request(path).post(Entity.entity(actions, MediaType.APPLICATION_JSON), ErrorData.class);
- } catch (Exception e) {
- throw getApiException(e);
+ grantPermissionAsync(topic, role, actions).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<Void> grantPermissionAsync(String topic, String role, Set<AuthAction> actions) {
+ TopicName tn = TopicName.get(topic);
+ WebTarget path = topicPath(tn, "permissions", role);
+ return asyncPostRequest(path, Entity.entity(actions, MediaType.APPLICATION_JSON));
+ }
+
+ @Override
public void revokePermissions(String topic, String role) throws PulsarAdminException {
try {
- TopicName tn = TopicName.get(topic);
- WebTarget path = topicPath(tn, "permissions", role);
- request(path).delete(ErrorData.class);
- } catch (Exception e) {
- throw getApiException(e);
+ revokePermissionsAsync(topic, role).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<Void> revokePermissionsAsync(String topic, String role) {
+ TopicName tn = TopicName.get(topic);
+ WebTarget path = topicPath(tn, "permissions", role);
+ return asyncDeleteRequest(path);
+ }
+
+ @Override
public void createPartitionedTopic(String topic, int numPartitions) throws PulsarAdminException {
try {
createPartitionedTopicAsync(topic, numPartitions).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
@@ -283,6 +388,21 @@ public class TopicsImpl extends BaseResource implements Topics {
}
@Override
+ public void updatePartitionedTopic(String topic, int numPartitions, boolean updateLocalTopicOnly) throws PulsarAdminException {
+ try {
+ updatePartitionedTopicAsync(topic, numPartitions, updateLocalTopicOnly)
+ .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<Void> updatePartitionedTopicAsync(String topic, int numPartitions,
boolean updateLocalTopicOnly) {
checkArgument(numPartitions > 0, "Number of partitions must be more than 0");
@@ -333,6 +453,11 @@ public class TopicsImpl extends BaseResource implements Topics {
}
@Override
+ public CompletableFuture<Void> deletePartitionedTopicAsync(String topic) {
+ return deletePartitionedTopicAsync(topic, false);
+ }
+
+ @Override
public void deletePartitionedTopic(String topic, boolean force) throws PulsarAdminException {
try {
deletePartitionedTopicAsync(topic, force).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
@@ -360,6 +485,11 @@ public class TopicsImpl extends BaseResource implements Topics {
}
@Override
+ public CompletableFuture<Void> deleteAsync(String topic) {
+ return deleteAsync(topic, false);
+ }
+
+ @Override
public void delete(String topic, boolean force) throws PulsarAdminException {
try {
deleteAsync(topic, force).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
@@ -403,6 +533,50 @@ public class TopicsImpl extends BaseResource implements Topics {
}
@Override
+ public MessageId terminateTopic(String topic) throws PulsarAdminException {
+ try {
+ return terminateTopicAsync(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<MessageId> terminateTopicAsync(String topic) {
+ TopicName tn = validateTopic(topic);
+
+ final CompletableFuture<MessageId> future = new CompletableFuture<>();
+ try {
+ final WebTarget path = topicPath(tn, "terminate");
+
+ request(path).async().post(Entity.entity("", MediaType.APPLICATION_JSON),
+ new InvocationCallback<MessageIdImpl>() {
+
+ @Override
+ public void completed(MessageIdImpl messageId) {
+ future.complete(messageId);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ log.warn("[{}] Failed to perform http post request: {}", path.getUri(),
+ throwable.getMessage());
+ future.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ } catch (PulsarAdminException cae) {
+ future.completeExceptionally(cae);
+ }
+
+ return future;
+ }
+
+ @Override
public List<String> getSubscriptions(String topic) throws PulsarAdminException {
try {
return getSubscriptionsAsync(topic).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
@@ -874,36 +1048,6 @@ public class TopicsImpl extends BaseResource implements Topics {
}
@Override
- public CompletableFuture<MessageId> terminateTopicAsync(String topic) {
- TopicName tn = validateTopic(topic);
-
- final CompletableFuture<MessageId> future = new CompletableFuture<>();
- try {
- final WebTarget path = topicPath(tn, "terminate");
-
- request(path).async().post(Entity.entity("", MediaType.APPLICATION_JSON),
- new InvocationCallback<MessageIdImpl>() {
-
- @Override
- public void completed(MessageIdImpl messageId) {
- future.complete(messageId);
- }
-
- @Override
- public void failed(Throwable throwable) {
- log.warn("[{}] Failed to perform http post request: {}", path.getUri(),
- throwable.getMessage());
- future.completeExceptionally(getApiException(throwable.getCause()));
- }
- });
- } catch (PulsarAdminException cae) {
- future.completeExceptionally(cae);
- }
-
- return future;
- }
-
- @Override
public void triggerCompaction(String topic) throws PulsarAdminException {
try {
triggerCompactionAsync(topic).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
@@ -928,37 +1072,110 @@ public class TopicsImpl extends BaseResource implements Topics {
public LongRunningProcessStatus compactionStatus(String topic)
throws PulsarAdminException {
try {
- TopicName tn = validateTopic(topic);
- return request(topicPath(tn, "compaction"))
- .get(LongRunningProcessStatus.class);
- } catch (Exception e) {
- throw getApiException(e);
+ return compactionStatusAsync(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<LongRunningProcessStatus> compactionStatusAsync(String topic) {
+ TopicName tn = validateTopic(topic);
+ WebTarget path = topicPath(tn, "compaction");
+ final CompletableFuture<LongRunningProcessStatus> future = new CompletableFuture<>();
+ asyncGetRequest(path,
+ new InvocationCallback<LongRunningProcessStatus>() {
+ @Override
+ public void completed(LongRunningProcessStatus longRunningProcessStatus) {
+ future.complete(longRunningProcessStatus);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ return future;
+ }
+
+ @Override
public void triggerOffload(String topic, MessageId messageId) throws PulsarAdminException {
try {
- TopicName tn = validateTopic(topic);
- WebTarget path = topicPath(tn, "offload");
- request(path).put(Entity.entity(messageId, MediaType.APPLICATION_JSON), MessageIdImpl.class);
- } catch (Exception e) {
- throw getApiException(e);
+ triggerOffloadAsync(topic, messageId).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<Void> triggerOffloadAsync(String topic, MessageId messageId) {
+ TopicName tn = validateTopic(topic);
+ WebTarget path = topicPath(tn, "offload");
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ try {
+ request(path).async().put(Entity.entity(messageId, MediaType.APPLICATION_JSON), new InvocationCallback<MessageIdImpl>() {
+ @Override
+ public void completed(MessageIdImpl response) {
+ future.complete(null);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(getApiException(throwable.getCause()));
+ }
+
+ });
+ } catch (PulsarAdminException cae) {
+ future.completeExceptionally(cae);
}
+ return future;
}
@Override
public OffloadProcessStatus offloadStatus(String topic)
throws PulsarAdminException {
try {
- TopicName tn = validateTopic(topic);
- return request(topicPath(tn, "offload"))
- .get(OffloadProcessStatus.class);
- } catch (Exception e) {
- throw getApiException(e);
+ return offloadStatusAsync(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<OffloadProcessStatus> offloadStatusAsync(String topic) {
+ TopicName tn = validateTopic(topic);
+ WebTarget path = topicPath(tn, "offload");
+ final CompletableFuture<OffloadProcessStatus> future = new CompletableFuture<>();
+ asyncGetRequest(path,
+ new InvocationCallback<OffloadProcessStatus>() {
+ @Override
+ public void completed(OffloadProcessStatus offloadProcessStatus) {
+ future.complete(offloadProcessStatus);
+ }
+
+ @Override
+ public void failed(Throwable throwable) {
+ future.completeExceptionally(getApiException(throwable.getCause()));
+ }
+ });
+ return future;
+ }
+
private WebTarget namespacePath(String domain, NamespaceName namespace, String... parts) {
final WebTarget base = namespace.isV2() ? adminV2Topics : adminTopics;
WebTarget namespacePath = base.path(domain).path(namespace.toString());
@@ -1058,6 +1275,7 @@ public class TopicsImpl extends BaseResource implements Topics {
}
}
+ @Override
public CompletableFuture<MessageId> getLastMessageIdAsync(String topic) {
TopicName tn = validateTopic(topic);
WebTarget path = topicPath(tn, "lastMessageId");