[pulsar] branch master updated: Bugfix an unload non-existent topic (#3946)

[zh...@apache.org]

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 e36ae70  Bugfix an unload non-existent topic (#3946)
e36ae70 is described below

commit e36ae702ea051913a723a7206e0bad7ae7e1949e
Author: Ezequiel Lovelle <ez...@gmail.com>
AuthorDate: Mon Apr 1 04:14:55 2019 -0300

    Bugfix an unload non-existent topic (#3946)
    
    *Motivation*
    
    Fixes #3935
    
    When an unload operation is being made to an non-existent topic a null pointer
    exception is thrown due to bad handling exception on general catch clause.
    
    *Modifications*
    
      - Refactor getTopicReference() for a simplified version.
      - Simplify exception thrown when partitioned topics was not found.
      - Fix NPE when unload operation is being made with a non-existent topic.
      - Add test in order to assert unload on normal situation.
      - Add test to assert npe from unload with non existent topic.
---
 .../broker/admin/impl/PersistentTopicsBase.java    | 41 +++++++++++-----------
 .../pulsar/broker/admin/PersistentTopicsTest.java  | 18 ++++++++++
 2 files changed, 39 insertions(+), 20 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 ce32890..c7af055 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
@@ -1309,24 +1309,25 @@ public class PersistentTopicsBase extends AdminResource {
      */
     private Topic getTopicReference(TopicName topicName) {
         return pulsar().getBrokerService().getTopicIfExists(topicName.toString()).join()
-                .orElseThrow(() -> {
-                    if (topicName.toString().contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) {
-                        TopicName partitionTopicName = TopicName.get(topicName.getPartitionedTopicName());
-                        PartitionedTopicMetadata partitionedTopicMetadata = getPartitionedTopicMetadata(partitionTopicName, false);
-                        if (partitionedTopicMetadata == null || partitionedTopicMetadata.partitions == 0) {
-                        	final String errSrc;
-                        	if (partitionedTopicMetadata != null) {
-                        		errSrc = " has zero partitions";
-                        	} else {
-                        		errSrc = " has no metadata";
-                        	}
-                            return new RestException(Status.NOT_FOUND, "Partitioned Topic not found: " + topicName.toString() + errSrc);
-                        } else if (!internalGetList().contains(topicName.toString())) {
-                            return new RestException(Status.NOT_FOUND, "Topic partitions were not yet created");
-                        }
-                    }
-                    return new RestException(Status.NOT_FOUND, "Topic not found");
-                });
+                .orElseThrow(() -> topicNotFoundReason(topicName));
+    }
+
+    private RestException topicNotFoundReason(TopicName topicName) {
+        if (!topicName.isPartitioned()) {
+            return new RestException(Status.NOT_FOUND, "Topic not found");
+        }
+
+        PartitionedTopicMetadata partitionedTopicMetadata = getPartitionedTopicMetadata(
+                TopicName.get(topicName.getPartitionedTopicName()), false);
+        if (partitionedTopicMetadata == null || partitionedTopicMetadata.partitions == 0) {
+            final String topicErrorType = partitionedTopicMetadata == null ?
+                    "has no metadata" : "has zero partitions";
+            return new RestException(Status.NOT_FOUND, String.format(
+                    "Partitioned Topic not found: %s %s", topicName.toString(), topicErrorType));
+        } else if (!internalGetList().contains(topicName.toString())) {
+            return new RestException(Status.NOT_FOUND, "Topic partitions were not yet created");
+        }
+        return new RestException(Status.NOT_FOUND, "Partitioned Topic not found");
     }
 
     private Topic getOrCreateTopic(TopicName topicName) {
@@ -1463,8 +1464,8 @@ public class PersistentTopicsBase extends AdminResource {
             log.error("[{}] topic {} not found", clientAppId(), topicName);
             throw new RestException(Status.NOT_FOUND, "Topic does not exist");
         } catch (Exception e) {
-            log.error("[{}] Failed to unload topic {}, {}", clientAppId(), topicName, e.getCause().getMessage(), e);
-            throw new RestException(e.getCause());
+            log.error("[{}] Failed to unload topic {}, {}", clientAppId(), topicName, e.getMessage(), e);
+            throw new RestException(e);
         }
     }
 
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 8311ec8..7900786 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
@@ -35,6 +35,7 @@ import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import javax.ws.rs.core.Response;
 import javax.ws.rs.core.UriInfo;
 import java.lang.reflect.Field;
 import java.util.List;
@@ -140,4 +141,21 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
                 testTenant, testNamespace, topicName, true);
         Assert.assertEquals(pMetadata.partitions, 0);
     }
+
+    @Test
+    public void testUnloadTopic() {
+        final String topicName = "standard-topic-to-be-unload";
+        persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, true);
+        persistentTopics.unloadTopic(testTenant, testNamespace, topicName, true);
+    }
+
+    @Test(expectedExceptions = RestException.class)
+    public void testUnloadTopicShallThrowNotFoundWhenTopicNotExist() {
+        try {
+            persistentTopics.unloadTopic(testTenant, testNamespace,"non-existent-topic", true);
+        } catch (RestException e) {
+            Assert.assertEquals(e.getResponse().getStatus(), Response.Status.NOT_FOUND.getStatusCode());
+            throw e;
+        }
+    }
 }