You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2022/12/07 08:04:13 UTC
[pulsar] branch branch-2.10 updated (6d86b0aa023 -> 012b85acaae)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a change to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 6d86b0aa023 [improve][admin] add topic name and sub name for NotFound error message (#15606)
new d21f842ff36 Revert "[improve][admin] add topic name and sub name for NotFound error message (#15606)"
new 012b85acaae Revert "PIP-105: new API to get subscription properties (#16095)"
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/pulsar/broker/admin/AdminResource.java | 8 -
.../broker/admin/impl/PersistentTopicsBase.java | 223 ++++-----------------
.../pulsar/broker/admin/v2/PersistentTopics.java | 36 ----
.../apache/pulsar/broker/admin/AdminApi2Test.java | 2 -
.../broker/admin/AdminApiSubscriptionTest.java | 48 +----
.../apache/pulsar/broker/admin/AdminApiTest.java | 31 +--
.../pulsar/broker/admin/PersistentTopicsTest.java | 3 +-
.../org/apache/pulsar/broker/admin/TopicsTest.java | 4 +-
.../org/apache/pulsar/client/admin/Topics.java | 16 --
.../pulsar/client/admin/internal/TopicsImpl.java | 29 ---
.../pulsar/admin/cli/PulsarAdminToolTest.java | 4 -
.../org/apache/pulsar/admin/cli/CmdTopics.java | 20 --
.../pulsar/tests/integration/cli/CLITest.java | 26 ---
13 files changed, 50 insertions(+), 400 deletions(-)
[pulsar] 01/02: Revert "[improve][admin] add topic name and sub name for NotFound error message (#15606)"
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d21f842ff36ae7f30a3768b74ad2a80d472c8ee0
Author: xiangying <19...@qq.com>
AuthorDate: Wed Dec 7 16:03:34 2022 +0800
Revert "[improve][admin] add topic name and sub name for NotFound error message (#15606)"
This reverts commit 6d86b0aa02345e4057aff86c1d970fd2e08848cf.
---
.../apache/pulsar/broker/admin/AdminResource.java | 8 --
.../broker/admin/impl/PersistentTopicsBase.java | 109 ++++++++-------------
.../apache/pulsar/broker/admin/AdminApi2Test.java | 2 -
.../broker/admin/AdminApiSubscriptionTest.java | 12 +--
.../apache/pulsar/broker/admin/AdminApiTest.java | 31 +-----
.../pulsar/broker/admin/PersistentTopicsTest.java | 3 +-
.../org/apache/pulsar/broker/admin/TopicsTest.java | 4 +-
7 files changed, 50 insertions(+), 119 deletions(-)
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 08ca6c7f23e..6b964fafe7e 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
@@ -839,12 +839,4 @@ public abstract class AdminResource extends PulsarWebResource {
protected static String getTopicNotFoundErrorMessage(String topic) {
return String.format("Topic %s not found", topic);
}
-
- protected static String getPartitionedTopicNotFoundErrorMessage(String topic) {
- return String.format("Partitioned Topic %s not found", topic);
- }
-
- protected static String getSubNotFoundErrorMessage(String topic, String subscription) {
- return String.format("Subscription %s not found for topic %s", subscription, topic);
- }
}
\ No newline at end of file
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 196586fec96..956679df95b 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
@@ -591,7 +591,7 @@ public class PersistentTopicsBase extends AdminResource {
return pulsar().getNamespaceService().checkTopicExists(topicName)
.thenAccept(exist -> {
if (!exist) {
- throw new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));
+ throw new RestException(Status.NOT_FOUND, "Topic not exist");
}
});
}
@@ -664,8 +664,7 @@ public class PersistentTopicsBase extends AdminResource {
} else if (realCause instanceof MetadataStoreException.NotFoundException) {
log.warn("Namespace policies of {} not found", topicName.getNamespaceObject());
asyncResponse.resume(new RestException(
- new RestException(Status.NOT_FOUND,
- getPartitionedTopicNotFoundErrorMessage(topicName.toString()))));
+ new RestException(Status.NOT_FOUND, "Partitioned topic does not exist")));
} else if (realCause instanceof PulsarAdminException) {
asyncResponse.resume(new RestException((PulsarAdminException) realCause));
} else if (realCause instanceof MetadataStoreException.BadVersionException) {
@@ -1081,7 +1080,7 @@ public class PersistentTopicsBase extends AdminResource {
if (t instanceof TopicBusyException) {
throw new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions");
} else if (isManagedLedgerNotFoundException(e)) {
- throw new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));
+ throw new RestException(Status.NOT_FOUND, "Topic not found");
} else {
throw new RestException(t);
}
@@ -1307,8 +1306,7 @@ public class PersistentTopicsBase extends AdminResource {
if (exception != null) {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
} else {
log.error("[{}] Failed to get managed info for {}", clientAppId(), topicName, t);
asyncResponse.resume(new RestException(t));
@@ -1380,8 +1378,7 @@ public class PersistentTopicsBase extends AdminResource {
future.thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName,
authoritative, false)).thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions == 0) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getPartitionedTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned Topic not found"));
return;
}
PartitionedTopicStatsImpl stats = new PartitionedTopicStatsImpl(partitionMetadata);
@@ -1455,8 +1452,7 @@ public class PersistentTopicsBase extends AdminResource {
future.thenCompose(__ -> getPartitionedTopicMetadataAsync(topicName, authoritative, false))
.thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions == 0) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getPartitionedTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Partitioned Topic not found"));
return;
}
@@ -1544,8 +1540,7 @@ public class PersistentTopicsBase extends AdminResource {
if (exception != null) {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return null;
} else if (t instanceof PreconditionFailedException) {
asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
@@ -1591,8 +1586,7 @@ public class PersistentTopicsBase extends AdminResource {
Topic topic = getTopicReference(topicName);
Subscription sub = topic.getSubscription(subName);
if (sub == null) {
- throw new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName));
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
}
return sub.delete();
}).thenRun(() -> {
@@ -1707,8 +1701,7 @@ public class PersistentTopicsBase extends AdminResource {
if (exception != null) {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return null;
} else {
log.error("[{}] Failed to delete subscription forcefully {} {}",
@@ -1754,8 +1747,7 @@ public class PersistentTopicsBase extends AdminResource {
Topic topic = getTopicReference(topicName);
Subscription sub = topic.getSubscription(subName);
if (sub == null) {
- throw new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName));
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
}
return sub.deleteForcefully();
}).thenRun(() -> {
@@ -1813,8 +1805,7 @@ public class PersistentTopicsBase extends AdminResource {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
asyncResponse.resume(
- new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ new RestException(Status.NOT_FOUND, "Subscription not found"));
} else {
log.error("[{}] Failed to skip all messages {} {}",
clientAppId(), topicName, subName, t);
@@ -1862,16 +1853,14 @@ public class PersistentTopicsBase extends AdminResource {
PersistentReplicator repl =
(PersistentReplicator) topic.getPersistentReplicator(remoteCluster);
if (repl == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return CompletableFuture.completedFuture(null);
}
return repl.clearBacklog().whenComplete(biConsumer);
} else {
PersistentSubscription sub = topic.getSubscription(subName);
if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return CompletableFuture.completedFuture(null);
}
return sub.clearBacklog().whenComplete(biConsumer);
@@ -1907,8 +1896,7 @@ public class PersistentTopicsBase extends AdminResource {
return getTopicReferenceAsync(topicName).thenCompose(t -> {
PersistentTopic topic = (PersistentTopic) t;
if (topic == null) {
- throw new RestException(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ throw new RestException(new RestException(Status.NOT_FOUND, "Topic not found"));
}
if (subName.startsWith(topic.getReplicatorPrefix())) {
String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
@@ -1928,8 +1916,7 @@ public class PersistentTopicsBase extends AdminResource {
PersistentSubscription sub = topic.getSubscription(subName);
if (sub == null) {
return FutureUtil.failedFuture(
- new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ new RestException(Status.NOT_FOUND, "Subscription not found"));
}
return sub.skipMessages(numMessages).thenAccept(unused -> {
log.info("[{}] Skipped {} messages on {} {}", clientAppId(), numMessages,
@@ -2030,7 +2017,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(__ -> getTopicReferenceAsync(topicName).thenAccept(t -> {
if (t == null) {
resumeAsyncResponseExceptionally(asyncResponse, new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ "Topic not found"));
return;
}
if (!(t instanceof PersistentTopic)) {
@@ -2176,7 +2163,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(topic -> {
Subscription sub = topic.getSubscription(subName);
if (sub == null) {
- throw new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
}
return sub.resetCursor(timestamp);
})
@@ -2537,14 +2524,12 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(ignore -> getTopicReferenceAsync(topicName))
.thenAccept(topic -> {
if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
PersistentSubscription sub = ((PersistentTopic) topic).getSubscription(subName);
if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return;
}
CompletableFuture<Integer> batchSizeFuture = new CompletableFuture<>();
@@ -3109,7 +3094,7 @@ public class PersistentTopicsBase extends AdminResource {
messageId.getEntryId());
if (topic == null) {
asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ "Topic not found"));
return;
}
ManagedLedgerImpl managedLedger =
@@ -3507,7 +3492,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(__ -> checkTopicExistsAsync(topicName))
.thenCompose(exist -> {
if (!exist) {
- throw new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));
+ throw new RestException(Status.NOT_FOUND, "Topic not found");
} else {
return getPartitionedTopicMetadataAsync(topicName, false, false)
.thenCompose(metadata -> {
@@ -3639,8 +3624,7 @@ public class PersistentTopicsBase extends AdminResource {
if (exception != null) {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
} else {
log.error("[{}] Failed to terminate topic {}", clientAppId(), topicName, t);
asyncResponse.resume(new RestException(t));
@@ -3716,8 +3700,7 @@ public class PersistentTopicsBase extends AdminResource {
Throwable t = exception.getCause();
if (t instanceof NotFoundException) {
asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(),
- subName)));
+ "Subscription not found"));
return null;
} else {
log.error("[{}] Failed to expire messages up "
@@ -3760,8 +3743,7 @@ public class PersistentTopicsBase extends AdminResource {
final CompletableFuture<Void> resultFuture = new CompletableFuture<>();
getTopicReferenceAsync(topicName).thenAccept(t -> {
if (t == null) {
- resultFuture.completeExceptionally(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ resultFuture.completeExceptionally(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
if (!(t instanceof PersistentTopic)) {
@@ -3786,8 +3768,7 @@ public class PersistentTopicsBase extends AdminResource {
PersistentSubscription sub = topic.getSubscription(subName);
if (sub == null) {
resultFuture.completeExceptionally(
- new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ new RestException(Status.NOT_FOUND, "Subscription not found"));
return;
}
issued = sub.expireMessages(expireTimeInSeconds);
@@ -3871,15 +3852,14 @@ public class PersistentTopicsBase extends AdminResource {
return getTopicReferenceAsync(topicName).thenAccept(t -> {
PersistentTopic topic = (PersistentTopic) t;
if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
try {
PersistentSubscription sub = topic.getSubscription(subName);
if (sub == null) {
asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ "Subscription not found"));
return;
}
CompletableFuture<Integer> batchSizeFuture = new CompletableFuture<>();
@@ -4208,7 +4188,7 @@ public class PersistentTopicsBase extends AdminResource {
private RestException topicNotFoundReason(TopicName topicName) {
if (!topicName.isPartitioned()) {
- return new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));
+ return new RestException(Status.NOT_FOUND, "Topic not found");
}
PartitionedTopicMetadata partitionedTopicMetadata = getPartitionedTopicMetadata(
@@ -4221,13 +4201,12 @@ public class PersistentTopicsBase extends AdminResource {
} else if (!internalGetList(Optional.empty()).contains(topicName.toString())) {
return new RestException(Status.NOT_FOUND, "Topic partitions were not yet created");
}
- return new RestException(Status.NOT_FOUND, getPartitionedTopicNotFoundErrorMessage(topicName.toString()));
+ return new RestException(Status.NOT_FOUND, "Partitioned Topic not found");
}
private CompletableFuture<Topic> topicNotFoundReasonAsync(TopicName topicName) {
if (!topicName.isPartitioned()) {
- return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ return FutureUtil.failedFuture(new RestException(Status.NOT_FOUND, "Topic not found"));
}
return getPartitionedTopicMetadataAsync(
@@ -4241,8 +4220,7 @@ public class PersistentTopicsBase extends AdminResource {
} else if (!internalGetList(Optional.empty()).contains(topicName.toString())) {
throw new RestException(Status.NOT_FOUND, "Topic partitions were not yet created");
}
- throw new RestException(Status.NOT_FOUND,
- getPartitionedTopicNotFoundErrorMessage(topicName.toString()));
+ throw new RestException(Status.NOT_FOUND, "Partitioned Topic not found");
});
}
@@ -4264,7 +4242,7 @@ public class PersistentTopicsBase extends AdminResource {
return checkNotNull(sub);
} catch (Exception e) {
- throw new RestException(Status.NOT_FOUND, getSubNotFoundErrorMessage(topicName.toString(), subName));
+ throw new RestException(Status.NOT_FOUND, "Subscription not found");
}
}
@@ -4542,8 +4520,7 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(__ -> getTopicReferenceAsync(topicName))
.thenAccept(topic -> {
if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
if (!(topic instanceof PersistentTopic)) {
@@ -5002,15 +4979,13 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(__ -> getTopicReferenceAsync(topicName))
.thenAccept(topic -> {
if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
Subscription sub = topic.getSubscription(subName);
if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
return;
}
@@ -5141,17 +5116,15 @@ public class PersistentTopicsBase extends AdminResource {
Topic topic = getTopicReference(topicName);
if (topic == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getTopicNotFoundErrorMessage(topicName.toString())));
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
return;
}
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
- return;
- }
+ Subscription sub = topic.getSubscription(subName);
+ if (sub == null) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Subscription not found"));
+ return;
+ }
if (topic instanceof PersistentTopic && sub instanceof PersistentSubscription) {
Map res = Maps.newHashMap();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
index 8399074886d..3cb67eada5e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
@@ -585,7 +585,6 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.topics().resetCursor(topicName + "invalid", "my-sub", messageId);
fail("It should have failed due to invalid topic name");
} catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains(topicName));
// Ok
}
@@ -594,7 +593,6 @@ public class AdminApi2Test extends MockedPulsarServiceBaseTest {
admin.topics().resetCursor(topicName, "invalid-sub", messageId);
fail("It should have failed due to invalid subscription name");
} catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains("invalid-sub"));
// Ok
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
index 411565521d9..521ef4df1c8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
@@ -72,18 +72,18 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
@Test
public void testExpireMessageWithNonExistTopicAndNonExistSub() {
String uuid = UUID.randomUUID().toString();
- String topic = "persistent://public/default/test-expire-messages-non-exist-topic-" + uuid;
+ String topic = "test-expire-messages-non-exist-topic-" + uuid;
String subscriptionName = "test-expire-messages-non-exist-sub-" + uuid;
PulsarAdminException exception = expectThrows(PulsarAdminException.class,
() -> admin.topics().expireMessages(topic, subscriptionName, 1));
assertEquals(exception.getStatusCode(), Response.Status.NOT_FOUND.getStatusCode());
- assertEquals(exception.getMessage(), String.format("Topic %s not found", topic));
+ assertEquals(exception.getMessage(), "Topic not found");
exception = expectThrows(PulsarAdminException.class,
() -> admin.topics().expireMessagesForAllSubscriptions(topic, 1));
assertEquals(exception.getStatusCode(), Response.Status.NOT_FOUND.getStatusCode());
- assertEquals(exception.getMessage(), String.format("Topic %s not found", topic));
+ assertEquals(exception.getMessage(), "Topic not found");
}
@Test
@@ -100,18 +100,18 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
@Test
public void tesSkipMessageWithNonExistTopicAndNotExistSub() {
String uuid = UUID.randomUUID().toString();
- String topic = "persistent://public/default/test-skip-messages-non-exist-topic-" + uuid;
+ String topic = "test-skip-messages-non-exist-topic-" + uuid;
String subscriptionName = "test-skip-messages-non-exist-sub-" + uuid;
PulsarAdminException exception = expectThrows(PulsarAdminException.class,
() -> admin.topics().skipMessages(topic, subscriptionName, 1));
assertEquals(exception.getStatusCode(), Response.Status.NOT_FOUND.getStatusCode());
- assertEquals(exception.getMessage(), String.format("Topic %s not found", topic));
+ assertEquals(exception.getMessage(), "Topic not found");
exception = expectThrows(PulsarAdminException.class,
() -> admin.topics().skipAllMessages(topic, subscriptionName));
assertEquals(exception.getStatusCode(), Response.Status.NOT_FOUND.getStatusCode());
- assertEquals(exception.getMessage(), String.format("Topic %s not found", topic));
+ assertEquals(exception.getMessage(), "Topic not found");
}
@DataProvider(name = "partitioned")
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index b2ebd2c1f6d..1c6e469935d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -892,7 +892,6 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
try {
admin.topics().skipAllMessages(persistentTopicName, subName);
} catch (NotFoundException e) {
- assertTrue(e.getMessage().contains(subName));
}
admin.topics().delete(persistentTopicName);
@@ -901,7 +900,6 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
admin.topics().delete(persistentTopicName);
fail("Should have received 404");
} catch (NotFoundException e) {
- assertTrue(e.getMessage().contains(persistentTopicName));
}
assertEquals(admin.topics().getList("prop-xyz/ns1"), Lists.newArrayList());
@@ -1102,7 +1100,6 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
admin.topics().deletePartitionedTopic(anotherTopic);
fail("Should have failed as the partitioned topic was not created");
} catch (NotFoundException nfe) {
- assertTrue(nfe.getMessage().contains(anotherTopic));
}
admin.topics().deletePartitionedTopic(partitionedTopicName);
@@ -1936,7 +1933,7 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
admin.topics().getStats("persistent://prop-xyz/ns1/ghostTopic");
fail("The topic doesn't exist");
} catch (NotFoundException e) {
- assertTrue(e.getMessage().contains("persistent://prop-xyz/ns1/ghostTopic"));
+ // OK
}
}
@@ -2026,20 +2023,6 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
topicName = "persistent://prop-xyz/ns1/" + topicName;
- try {
- admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis());
- } catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains(topicName));
- }
-
- admin.topics().createNonPartitionedTopic(topicName);
-
- try {
- admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis());
- } catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains("my-sub"));
- }
-
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
.subscriptionName("my-sub").startMessageIdInclusive()
@@ -2232,20 +2215,8 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
admin.namespaces().setRetention("prop-xyz/ns1", new RetentionPolicies(10, 10));
topicName = "persistent://prop-xyz/ns1/" + topicName;
- try {
- admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis());
- } catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains(topicName));
- }
-
admin.topics().createPartitionedTopic(topicName, 4);
- try {
- admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis());
- } catch (PulsarAdminException.NotFoundException e) {
- assertTrue(e.getMessage().contains("my-sub"));
- }
-
// create consumer and subscription
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
.subscriptionName("my-sub").startMessageIdInclusive()
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 44de487113d..bd86a666c0f 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
@@ -172,8 +172,7 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
Assert.assertEquals(errorCaptor.getValue().getResponse().getStatus(),
Response.Status.NOT_FOUND.getStatusCode());
- Assert.assertEquals(errorCaptor.getValue().getMessage(), String.format("Topic %s not found",
- "persistent://my-tenant/my-namespace/topic-not-found"));
+ Assert.assertEquals(errorCaptor.getValue().getMessage(), "Topic not found");
// 2) Confirm that the partitioned topic does not exist
response = mock(AsyncResponse.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
index 3327b83b6ae..35e205f2dc9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
@@ -383,9 +383,7 @@ public class TopicsTest extends MockedPulsarServiceBaseTest {
topics.produceOnPersistentTopic(asyncResponse, testTenant, testNamespace, testTopicName, false, producerMessages);
ArgumentCaptor<RestException> responseCaptor = ArgumentCaptor.forClass(RestException.class);
verify(asyncResponse, timeout(5000).times(1)).resume(responseCaptor.capture());
- System.out.println(responseCaptor.getValue().getMessage());
- Assert.assertTrue(responseCaptor.getValue().getMessage()
- .contains(String.format("Topic %s not found", topicName)));
+ Assert.assertTrue(responseCaptor.getValue().getMessage().contains("Topic not exist"));
}
@Test
[pulsar] 02/02: Revert "PIP-105: new API to get subscription properties (#16095)"
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 012b85acaae974ee451f36560ce1a1295fb3b221
Author: xiangying <19...@qq.com>
AuthorDate: Wed Dec 7 16:03:47 2022 +0800
Revert "PIP-105: new API to get subscription properties (#16095)"
This reverts commit d2ff2936097f8920b4521923685b7623eda5fcdc.
---
.../broker/admin/impl/PersistentTopicsBase.java | 114 ---------------------
.../pulsar/broker/admin/v2/PersistentTopics.java | 36 -------
.../broker/admin/AdminApiSubscriptionTest.java | 36 -------
.../org/apache/pulsar/client/admin/Topics.java | 16 ---
.../pulsar/client/admin/internal/TopicsImpl.java | 29 ------
.../pulsar/admin/cli/PulsarAdminToolTest.java | 4 -
.../org/apache/pulsar/admin/cli/CmdTopics.java | 20 ----
.../pulsar/tests/integration/cli/CLITest.java | 26 -----
8 files changed, 281 deletions(-)
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 956679df95b..698da80265d 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
@@ -1636,35 +1636,6 @@ public class PersistentTopicsBase extends AdminResource {
});
}
- private void internalGetSubscriptionPropertiesForNonPartitionedTopic(AsyncResponse asyncResponse,
- String subName,
- boolean authoritative) {
- validateTopicOwnershipAsync(topicName, authoritative)
- .thenRun(() -> validateTopicOperation(topicName, TopicOperation.CONSUME))
- .thenCompose(__ -> getTopicReferenceAsync(topicName))
- .thenApply((Topic topic) -> {
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- throw new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName));
- }
- return sub.getSubscriptionProperties();
- }).thenAccept((Map<String, String> properties) -> {
- if (properties == null) {
- properties = Collections.emptyMap();
- }
- asyncResponse.resume(Response.ok(properties).build());
- }).exceptionally(ex -> {
- Throwable cause = ex.getCause();
- // If the exception is not redirect exception we need to log it.
- if (!isRedirectException(ex)) {
- log.error("[{}] Failed to update subscription {} {}", clientAppId(), topicName, subName, cause);
- }
- asyncResponse.resume(new RestException(cause));
- return null;
- });
- }
-
protected void internalDeleteSubscriptionForcefully(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
CompletableFuture<Void> future;
@@ -2413,91 +2384,6 @@ public class PersistentTopicsBase extends AdminResource {
});
}
- protected void internalGetSubscriptionProperties(AsyncResponse asyncResponse, String subName,
- boolean authoritative) {
- CompletableFuture<Void> future;
- if (topicName.isGlobal()) {
- future = validateGlobalNamespaceOwnershipAsync(namespaceName);
- } else {
- future = CompletableFuture.completedFuture(null);
- }
-
- future.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative)).thenAccept(__ -> {
- if (topicName.isPartitioned()) {
- internalGetSubscriptionPropertiesForNonPartitionedTopic(asyncResponse, subName,
- authoritative);
- } else {
- getPartitionedTopicMetadataAsync(topicName,
- authoritative, false).thenAcceptAsync(partitionMetadata -> {
- if (partitionMetadata.partitions > 0) {
- final List<CompletableFuture<Map<String, String>>> futures = Lists.newArrayList();
-
- for (int i = 0; i < partitionMetadata.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics()
- .getSubscriptionPropertiesAsync(topicNamePartition.toString(),
- subName));
- } catch (Exception e) {
- log.error("[{}] Failed to update properties for subscription {} {}",
- clientAppId(), topicNamePartition, subName,
- e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable t = exception.getCause();
- if (t instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND,
- getSubNotFoundErrorMessage(topicName.toString(), subName)));
- return null;
- } else {
- log.error("[{}] Failed to get properties for subscription {} {}",
- clientAppId(), topicName, subName, t);
- asyncResponse.resume(new RestException(t));
- return null;
- }
- }
-
- Map<String, String> aggregatedResult = new HashMap<>();
- futures.forEach(f -> {
- // in theory all the partitions have the same properties
- try {
- aggregatedResult.putAll(f.get());
- } catch (Exception impossible) {
- // we already waited for this Future
- asyncResponse.resume(new RestException(impossible));
- }
- });
-
- asyncResponse.resume(Response.ok(aggregatedResult).build());
- return null;
- });
- } else {
- internalGetSubscriptionPropertiesForNonPartitionedTopic(asyncResponse, subName,
- authoritative);
- }
- }, pulsar().getExecutor()).exceptionally(ex -> {
- log.error("[{}] Failed to update properties for subscription {} from topic {}",
- clientAppId(), subName, topicName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
- }
- }).exceptionally(ex -> {
- // If the exception is not redirect exception we need to log it.
- if (!isRedirectException(ex)) {
- log.error("[{}] Failed to update subscription {} from topic {}",
- clientAppId(), subName, topicName, ex);
- }
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
- }
-
protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String subName, boolean authoritative,
MessageIdImpl messageId, boolean isExcluded, int batchIndex) {
CompletableFuture<Void> ret;
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 00cc19e0c98..56e3799c475 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
@@ -1564,42 +1564,6 @@ public class PersistentTopics extends PersistentTopicsBase {
}
}
- @GET
- @Path("/{tenant}/{namespace}/{topic}/subscription/{subName}/properties")
- @ApiOperation(value = "Replaces all the properties on the given subscription")
- @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 or"
- + "subscriber is not authorized to access this operation"),
- @ApiResponse(code = 403, message = "Don't have admin permission"),
- @ApiResponse(code = 404, message = "Topic/Subscription does not exist"),
- @ApiResponse(code = 405, message = "Method Not Allowed"),
- @ApiResponse(code = 500, message = "Internal server error"),
- @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
- })
- public void getSubscriptionProperties(
- @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 = "Subscription", required = true)
- @PathParam("subName") String encodedSubName,
- @ApiParam(value = "Is authentication required to perform this operation")
- @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateTopicName(tenant, namespace, encodedTopic);
- internalGetSubscriptionProperties(asyncResponse, decode(encodedSubName),
- authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
- }
-
@POST
@Path("/{tenant}/{namespace}/{topic}/subscription/{subName}/resetcursor")
@ApiOperation(value = "Reset subscription to message position closest to given position.",
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
index 521ef4df1c8..f7af28ddf41 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSubscriptionTest.java
@@ -152,9 +152,6 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic + "-partition-" + i)
.getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
-
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic + "-partition-" + i, subscriptionName);
- assertEquals(value, props.get("foo"));
}
// properties are never null, but an empty map
@@ -162,9 +159,6 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic + "-partition-" + i)
.getSubscriptions().get(subscriptionName2);
assertTrue(subscriptionStats.getSubscriptionProperties().isEmpty());
-
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic + "-partition-" + i, subscriptionName2);
- assertTrue(props.isEmpty());
}
// aggregated properties
@@ -172,21 +166,12 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
.getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertEquals(value, props.get("foo"));
-
} else {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic).getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertEquals(value, props.get("foo"));
-
SubscriptionStats subscriptionStats2 = admin.topics().getStats(topic).getSubscriptions().get(subscriptionName2);
assertTrue(subscriptionStats2.getSubscriptionProperties().isEmpty());
-
- Map<String, String> props2 = admin.topics().getSubscriptionProperties(topic, subscriptionName2);
- assertTrue(props2.isEmpty());
}
// clear the properties on subscriptionName
@@ -198,9 +183,6 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic + "-partition-" + i)
.getSubscriptions().get(subscriptionName);
assertTrue(subscriptionStats.getSubscriptionProperties().isEmpty());
-
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic + "-partition-" + i, subscriptionName);
- assertTrue(props.isEmpty());
}
// aggregated properties
@@ -208,15 +190,9 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
.getSubscriptions().get(subscriptionName);
assertTrue(subscriptionStats.getSubscriptionProperties().isEmpty());
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertTrue(props.isEmpty());
-
} else {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic).getSubscriptions().get(subscriptionName);
assertTrue(subscriptionStats.getSubscriptionProperties().isEmpty());
-
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertTrue(props.isEmpty());
}
// update the properties on subscriptionName
@@ -228,9 +204,6 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic + "-partition-" + i)
.getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
-
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic + "-partition-" + i, subscriptionName);
- assertEquals(value, props.get("foo"));
}
// aggregated properties
@@ -238,21 +211,12 @@ public class AdminApiSubscriptionTest extends MockedPulsarServiceBaseTest {
.getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertEquals(value, props.get("foo"));
-
} else {
SubscriptionStats subscriptionStats = admin.topics().getStats(topic).getSubscriptions().get(subscriptionName);
assertEquals(value, subscriptionStats.getSubscriptionProperties().get("foo"));
- Map<String, String> props = admin.topics().getSubscriptionProperties(topic, subscriptionName);
- assertEquals(value, props.get("foo"));
-
SubscriptionStats subscriptionStats2 = admin.topics().getStats(topic).getSubscriptions().get(subscriptionName2);
assertTrue(subscriptionStats2.getSubscriptionProperties().isEmpty());
-
- Map<String, String> props2 = admin.topics().getSubscriptionProperties(topic, subscriptionName2);
- assertTrue(props2.isEmpty());
}
}
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 2fa14e54434..73f9a199a1b 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
@@ -1764,15 +1764,6 @@ public interface Topics {
void updateSubscriptionProperties(String topic, String subName, Map<String, String> subscriptionProperties)
throws PulsarAdminException;
- /**
- * Get Subscription Properties on a topic subscription.
- * @param topic
- * @param subName
- * @throws PulsarAdminException
- */
- Map<String, String> getSubscriptionProperties(String topic, String subName)
- throws PulsarAdminException;
-
/**
* Reset cursor position on a topic subscription.
*
@@ -1807,13 +1798,6 @@ public interface Topics {
CompletableFuture<Void> updateSubscriptionPropertiesAsync(String topic, String subName,
Map<String, String> subscriptionProperties);
- /**
- * Get Subscription Properties on a topic subscription.
- * @param topic
- * @param subName
- */
- CompletableFuture<Map<String, String>> getSubscriptionPropertiesAsync(String topic, String subName);
-
/**
* Reset cursor position on a topic subscription.
*
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 a2360fb5b73..dbdaffd9e5c 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
@@ -1179,12 +1179,6 @@ public class TopicsImpl extends BaseResource implements Topics {
sync(() -> updateSubscriptionPropertiesAsync(topic, subName, subscriptionProperties));
}
- @Override
- public Map<String, String> getSubscriptionProperties(String topic, String subName)
- throws PulsarAdminException {
- return sync(() -> getSubscriptionPropertiesAsync(topic, subName));
- }
-
@Override
public CompletableFuture<Void> updateSubscriptionPropertiesAsync(String topic, String subName,
Map<String, String> subscriptionProperties) {
@@ -1198,29 +1192,6 @@ public class TopicsImpl extends BaseResource implements Topics {
return asyncPutRequest(path, Entity.entity(subscriptionProperties, MediaType.APPLICATION_JSON));
}
- @Override
- public CompletableFuture<Map<String, String>> getSubscriptionPropertiesAsync(String topic, String subName) {
- TopicName tn = validateTopic(topic);
- String encodedSubName = Codec.encode(subName);
- WebTarget path = topicPath(tn, "subscription", encodedSubName,
- "properties");
- final CompletableFuture<Map<String, String>> future = new CompletableFuture<>();
- asyncGetRequest(path,
- new InvocationCallback<Map<String, String>>() {
-
- @Override
- public void completed(Map<String, String> response) {
- future.complete(response);
- }
-
- @Override
- public void failed(Throwable throwable) {
- future.completeExceptionally(getApiException(throwable.getCause()));
- }
- });
- return future;
- }
-
@Override
public void resetCursor(String topic, String subName, MessageId messageId
, boolean isExcluded) throws PulsarAdminException {
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 067e08b5599..8e7e339b536 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
@@ -1422,10 +1422,6 @@ public class PulsarAdminToolTest {
cmdTopics.run(split("update-subscription-properties persistent://myprop/clust/ns1/ds1 -s sub1 --clear"));
verify(mockTopics).updateSubscriptionProperties("persistent://myprop/clust/ns1/ds1", "sub1", new HashMap<>());
- cmdTopics = new CmdTopics(() -> admin);
- cmdTopics.run(split("get-subscription-properties persistent://myprop/clust/ns1/ds1 -s sub1"));
- verify(mockTopics).getSubscriptionProperties("persistent://myprop/clust/ns1/ds1", "sub1");
-
cmdTopics = new CmdTopics(() -> admin);
props = new HashMap<>();
props.put("a", "b");
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 562d7993f5c..8fc8d5646d7 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
@@ -74,7 +74,6 @@ import org.apache.pulsar.common.policies.data.PublishRate;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.util.DateFormatter;
-import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.RelativeTimeUtil;
@Getter
@@ -100,7 +99,6 @@ public class CmdTopics extends CmdBase {
jcommander.addCommand("unsubscribe", new DeleteSubscription());
jcommander.addCommand("create-subscription", new CreateSubscription());
jcommander.addCommand("update-subscription-properties", new UpdateSubscriptionProperties());
- jcommander.addCommand("get-subscription-properties", new GetSubscriptionProperties());
jcommander.addCommand("stats", new GetStats());
jcommander.addCommand("stats-internal", new GetInternalStats());
@@ -970,24 +968,6 @@ public class CmdTopics extends CmdBase {
}
}
- @Parameters(commandDescription = "Get the properties of a subscription on a topic")
- private class GetSubscriptionProperties extends CliCommand {
- @Parameter(description = "persistent://tenant/namespace/topic", required = true)
- private java.util.List<String> params;
-
- @Parameter(names = { "-s",
- "--subscription" }, description = "Subscription to describe", required = true)
- private String subscriptionName;
-
- @Override
- void run() throws Exception {
- String topic = validateTopicName(params);
- Map<String, String> result = getTopics().getSubscriptionProperties(topic, subscriptionName);
- // Ensure we are using JSON and not Java toString()
- System.out.println(ObjectMapperFactory.getThreadLocal().writeValueAsString(result));
- }
- }
-
@Parameters(commandDescription = "Reset position for subscription to a position that is closest to "
+ "timestamp or messageId.")
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java
index cda2347b4dc..a1e417ca547 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java
@@ -207,18 +207,6 @@ public class CLITest extends PulsarTestSuite {
);
resultUpdate.assertNoOutput();
- ContainerExecResult resultGet = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "topics",
- "get-subscription-properties",
- "persistent://public/default/" + topic,
- "--subscription",
- "" + subscriptionPrefix + i
- );
- assertEquals(
- resultGet.getStdout().trim(), "{\"a\":\"e\"}",
- "unexpected output " + resultGet.getStdout() + " - error " + resultGet.getStderr());
-
ContainerExecResult resultClear = container.execCmd(
PulsarCluster.ADMIN_SCRIPT,
"topics",
@@ -229,20 +217,6 @@ public class CLITest extends PulsarTestSuite {
"" + subscriptionPrefix + i
);
resultClear.assertNoOutput();
-
- ContainerExecResult resultGetAfterClear = container.execCmd(
- PulsarCluster.ADMIN_SCRIPT,
- "topics",
- "get-subscription-properties",
- "persistent://public/default/" + topic,
- "--subscription",
- "" + subscriptionPrefix + i
- );
- assertEquals(
- resultGetAfterClear.getStdout().trim(), "{}",
- "unexpected output " + resultGetAfterClear.getStdout()
- + " - error " + resultGetAfterClear.getStderr());
-
i++;
}
}