[fix][broker] Terminate the async call chain when the condition isn't met for resetCursor (#19541)

Author: lhotari

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

@@ -2672,7 +2672,7 @@ protected void internalResetCursorOnPosition(AsyncResponse asyncResponse, String
                         if (topicMetadata.partitions > 0) {
                             log.warn("[{}] Not supported operation on partitioned-topic {} {}",
                                     clientAppId(), topicName, subName);
-                            asyncResponse.resume(new RestException(Status.METHOD_NOT_ALLOWED,
+                            throw new CompletionException(new RestException(Status.METHOD_NOT_ALLOWED,
                                     "Reset-cursor at position is not allowed for partitioned-topic"));
                         }
                         return CompletableFuture.completedFuture(null);

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java

@@ -661,6 +661,22 @@ public void testResetCursorOnPosition(String namespaceName) throws Exception {
         setup();
     }
 
+    @Test
+    public void shouldNotSupportResetOnPartitionedTopic() throws PulsarAdminException, PulsarClientException {
+        final String partitionedTopicName = "persistent://prop-xyz/ns1/" + BrokerTestUtil.newUniqueName("parttopic");
+        admin.topics().createPartitionedTopic(partitionedTopicName, 4);
+        @Cleanup
+        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(partitionedTopicName).subscriptionName("my-sub")
+                .subscriptionType(SubscriptionType.Shared).subscribe();
+        try {
+            admin.topics().resetCursor(partitionedTopicName, "my-sub", MessageId.earliest);
+            fail();
+        } catch (PulsarAdminException.NotAllowedException e) {
+            assertTrue(e.getMessage().contains("Reset-cursor at position is not allowed for partitioned-topic"),
+                    "Condition doesn't match. Actual message:" + e.getMessage());
+        }
+    }
+
     private void publishMessagesOnPersistentTopic(String topicName, int messages, int startIdx) throws Exception {
         Producer<byte[]> producer = pulsarClient.newProducer()
             .topic(topicName)