fix: fix non-persistent topic impl bug (#22)

Signed-off-by: ZhangJian He <[email protected]>
protocol-laboratory · Aug 28, 2024 · 6525edb · 6525edb
1 parent f0dbc40
commit 6525edb
Show file tree

Hide file tree

Showing 48 changed files with 112 additions and 105 deletions.
diff --git a/pom.xml b/pom.xml
@@ -24,11 +24,13 @@
         <jackson.version>2.17.2</jackson.version>
         <junit.version>5.11.0</junit.version>
         <log4j.version>2.20.0</log4j.version>
+        <lombok.version>1.18.34</lombok.version>
         <!-- plugin -->
         <compiler-plugin.version>3.10.1</compiler-plugin.version>
         <checkstyle-plugin.version>3.2.0</checkstyle-plugin.version>
         <javadoc-plugin.version>3.4.1</javadoc-plugin.version>
-        <lombok.version>1.18.34</lombok.version>
+        <lombok-maven-plugin.version>1.18.20.0</lombok-maven-plugin.version>
+        <maven-gpg-plugin.version>3.2.4</maven-gpg-plugin.version>
         <nexus-staging-maven-plugin.version>1.6.13</nexus-staging-maven-plugin.version>
         <puppycrawl.version>10.18.0</puppycrawl.version>
         <release-plugin.version>3.0.0-M6</release-plugin.version>
@@ -243,7 +245,7 @@
 
     <developers>
         <developer>
-            <name>ShootHzj</name>
+            <name>shootHzj</name>
             <email>[email protected]</email>
         </developer>
     </developers>
@@ -280,7 +282,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-gpg-plugin</artifactId>
-                        <version>3.0.1</version>
+                        <version>${maven-gpg-plugin.version}</version>
                         <executions>
                             <execution>
                                 <id>sign-artifacts</id>
@@ -311,12 +313,12 @@
                     <plugin>
                         <groupId>org.projectlombok</groupId>
                         <artifactId>lombok-maven-plugin</artifactId>
-                        <version>1.18.20.0</version>
+                        <version>${lombok-maven-plugin.version}</version>
                         <dependencies>
                             <dependency>
                                 <groupId>org.projectlombok</groupId>
                                 <artifactId>lombok</artifactId>
-                                <version>1.18.24</version>
+                                <version>${lombok.version}</version>
                             </dependency>
                         </dependencies>
                         <executions>

diff --git a/pulsar-admin-jdk/pom.xml b/pulsar-admin-jdk/pom.xml
@@ -11,4 +11,15 @@
 
     <artifactId>pulsar-admin-jdk</artifactId>
 
+    <build>
+        <testResources>
+            <testResource>
+                <directory>src/test/resources</directory>
+                <includes>
+                    <include>**.jks</include>
+                </includes>
+            </testResource>
+        </testResources>
+    </build>
+
 </project>
diff --git a/pulsar-admin-jdk/src/main/java/io/github/protocol/pulsar/NonPersistentTopicsImpl.java b/pulsar-admin-jdk/src/main/java/io/github/protocol/pulsar/NonPersistentTopicsImpl.java
diff --git a/.../github/protocol/pulsar/BacklogQuota.java → ...otocol/pulsar/admin/jdk/BacklogQuota.java b/.../github/protocol/pulsar/BacklogQuota.java → ...otocol/pulsar/admin/jdk/BacklogQuota.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

diff --git a/...hub/protocol/pulsar/BacklogQuotaType.java → ...ol/pulsar/admin/jdk/BacklogQuotaType.java b/...hub/protocol/pulsar/BacklogQuotaType.java → ...ol/pulsar/admin/jdk/BacklogQuotaType.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 public enum BacklogQuotaType {
 

diff --git a/...protocol/pulsar/PersistentTopicsImpl.java → ...ocol/pulsar/admin/jdk/BaseTopicsImpl.java b/...protocol/pulsar/PersistentTopicsImpl.java → ...ocol/pulsar/admin/jdk/BaseTopicsImpl.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 
@@ -7,25 +7,20 @@
 import java.util.List;
 import java.util.Map;
 
-public class PersistentTopicsImpl implements Topics {
+public abstract class BaseTopicsImpl implements Topics {
 
     protected final InnerHttpClient httpClient;
 
-    private static final String BASE_URL_PERSISTENT_DOMAIN = "/admin/v2" + "/persistent";
-
-    public PersistentTopicsImpl(InnerHttpClient httpClient) {
+    public BaseTopicsImpl(InnerHttpClient httpClient) {
         this.httpClient = httpClient;
     }
 
-    public String getDomainBaseUrl() {
-        return "/admin/v2" + "/persistent";
-    }
+    protected abstract String getDomainBaseUrl();
 
     @Override
     public void createPartitionedTopic(String tenant, String namespace, String encodedTopic, int numPartitions,
                                        boolean createLocalTopicOnly) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
-                "/partitions");
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic, "/partitions");
         try {
             HttpResponse<String> response = httpClient.put(url, numPartitions, "createLocalTopicOnly",
                     String.valueOf(createLocalTopicOnly));
@@ -42,8 +37,7 @@ public void createPartitionedTopic(String tenant, String namespace, String encod
     @Override
     public void deletePartitionedTopic(String tenant, String namespace, String encodedTopic, boolean force,
                                        boolean authoritative) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
-                "/partitions");
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic, "/partitions");
         try {
             HttpResponse<String> response = httpClient.delete(url, "force", String.valueOf(force),
                     "authoritative", String.valueOf(authoritative));
@@ -61,8 +55,7 @@ public void deletePartitionedTopic(String tenant, String namespace, String encod
     public void updatePartitionedTopic(String tenant, String namespace, String encodedTopic,
                                        boolean updateLocalTopicOnly, boolean authoritative,
                                        boolean force, int numPartitions) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
-                "/partitions");
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic, "/partitions");
         try {
             HttpResponse<String> response = httpClient.post(url, numPartitions, "updateLocalTopicOnly",
                     String.valueOf(updateLocalTopicOnly),
@@ -82,8 +75,7 @@ public void updatePartitionedTopic(String tenant, String namespace, String encod
     public PartitionedTopicMetadata getPartitionedMetadata(String tenant, String namespace, String encodedTopic,
                                                            boolean checkAllowAutoCreation, boolean authoritative)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent",
-                tenant, namespace, encodedTopic, "/partitions");
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic, "/partitions");
         try {
             HttpResponse<String> response = httpClient.get(url,
                     "checkAllowAutoCreation", String.valueOf(checkAllowAutoCreation),
@@ -102,7 +94,7 @@ public PartitionedTopicMetadata getPartitionedMetadata(String tenant, String nam
     @Override
     public void createNonPartitionedTopic(String tenant, String namespace, String encodedTopic, boolean authoritative,
                                           Map<String, String> properties) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic);
+        String url = String.format("%s/%s/%s/%s", getDomainBaseUrl(), tenant, namespace, encodedTopic);
         try {
             HttpResponse<String> response = httpClient.put(url, properties,
                     "authoritative", String.valueOf(authoritative));
@@ -119,7 +111,7 @@ public void createNonPartitionedTopic(String tenant, String namespace, String en
     @Override
     public void deleteTopic(String tenant, String namespace, String encodedTopic, boolean force, boolean authoritative)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic);
+        String url = String.format("%s/%s/%s/%s", getDomainBaseUrl(), tenant, namespace, encodedTopic);
         try {
             HttpResponse<String> response = httpClient.delete(url,
                     "force", String.valueOf(force),
@@ -137,7 +129,7 @@ public void deleteTopic(String tenant, String namespace, String encodedTopic, bo
     @Override
     public List<String> getList(String tenant, String namespace, String bundle, boolean includeSystemTopic)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s", "/admin/v2" + "/persistent", tenant, namespace);
+        String url = String.format("%s/%s/%s", getDomainBaseUrl(), tenant, namespace);
         try {
             HttpResponse<String> response;
             if (bundle != null) {
@@ -164,7 +156,7 @@ public List<String> getList(String tenant, String namespace, String bundle, bool
     @Override
     public List<String> getPartitionedTopicList(String tenant, String namespace, boolean includeSystemTopic)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, UrlConst.PARTITIONED);
+        String url = String.format("%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, UrlConst.PARTITIONED);
         try {
             HttpResponse<String> response = httpClient.get(url, "includeSystemTopic",
                     String.valueOf(includeSystemTopic));
@@ -184,7 +176,7 @@ public List<String> getPartitionedTopicList(String tenant, String namespace, boo
     @Override
     public void createMissedPartitions(String tenant, String namespace, String encodedTopic)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.CREATE_MISSED_PARTITIONS);
         try {
             HttpResponse<String> response = httpClient.post(url);
@@ -202,7 +194,7 @@ public void createMissedPartitions(String tenant, String namespace, String encod
     @Override
     public MessageIdImpl getLastMessageId(String tenant, String namespace, String encodedTopic, boolean authoritative)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.LAST_MESSAGE_ID);
         try {
             HttpResponse<String> response = httpClient.get(url, "authoritative", String.valueOf(authoritative));
@@ -220,7 +212,7 @@ public MessageIdImpl getLastMessageId(String tenant, String namespace, String en
     @Override
     public RetentionPolicies getRetention(String tenant, String namespace, String encodedTopic, boolean isGlobal,
                                           boolean applied, boolean authoritative) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.RETENTION);
         try {
             HttpResponse<String> response = httpClient.get(url,
@@ -241,7 +233,7 @@ public RetentionPolicies getRetention(String tenant, String namespace, String en
     @Override
     public void setRetention(String tenant, String namespace, String encodedTopic, boolean authoritative,
                              boolean isGlobal, RetentionPolicies retention) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.RETENTION);
         try {
             HttpResponse<String> response = httpClient.post(url, retention,
@@ -260,7 +252,7 @@ public void setRetention(String tenant, String namespace, String encodedTopic, b
     @Override
     public void removeRetention(String tenant, String namespace, String encodedTopic, boolean authoritative)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.RETENTION);
         try {
             HttpResponse<String> response = httpClient.delete(url, "authoritative", String.valueOf(authoritative));
@@ -279,7 +271,7 @@ public Map<BacklogQuotaType, BacklogQuota> getBacklogQuotaMap(String tenant, Str
                                                                   String encodedTopic, boolean applied,
                                                                   boolean authoritative, boolean isGlobal)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.BACKLOG_QUOTA_MAP);
         try {
             HttpResponse<String> response = httpClient.get(url,
@@ -302,7 +294,7 @@ public Map<BacklogQuotaType, BacklogQuota> getBacklogQuotaMap(String tenant, Str
     public void setBacklogQuota(String tenant, String namespace, String encodedTopic, boolean authoritative,
                                 boolean isGlobal, BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.BACKLOG_QUOTA);
         try {
             HttpResponse<String> response = httpClient.post(url, backlogQuota,
@@ -323,7 +315,7 @@ public void setBacklogQuota(String tenant, String namespace, String encodedTopic
     public void removeBacklogQuota(String tenant, String namespace, String encodedTopic,
                                    BacklogQuotaType backlogQuotaType, boolean authoritative, boolean isGlobal)
             throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.BACKLOG_QUOTA);
         try {
             HttpResponse<String> response = httpClient.delete(url,
@@ -343,7 +335,7 @@ public void removeBacklogQuota(String tenant, String namespace, String encodedTo
     @Override
     public PersistentOfflineTopicStats getBacklog(String tenant, String namespace, String encodedTopic,
                                                   boolean authoritative) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.BACKLOG);
         try {
             HttpResponse<String> response = httpClient.get(url, "authoritative", String.valueOf(authoritative));
@@ -361,7 +353,7 @@ public PersistentOfflineTopicStats getBacklog(String tenant, String namespace, S
     @Override
     public long getBacklogSizeByMessageId(String tenant, String namespace, String encodedTopic, boolean authoritative,
                                           MessageIdImpl messageId) throws PulsarAdminException {
-        String url = String.format("%s/%s/%s/%s%s", "/admin/v2" + "/persistent", tenant, namespace, encodedTopic,
+        String url = String.format("%s/%s/%s/%s%s", getDomainBaseUrl(), tenant, namespace, encodedTopic,
                 UrlConst.BACKLOG_SIZE);
         try {
             HttpResponse<String> response = httpClient.put(url, messageId, "authoritative",
@@ -376,5 +368,4 @@ public long getBacklogSizeByMessageId(String tenant, String namespace, String en
             throw new PulsarAdminException(e);
         }
     }
-
 }
diff --git a/...va/io/github/protocol/pulsar/Brokers.java → ...ub/protocol/pulsar/admin/jdk/Brokers.java b/...va/io/github/protocol/pulsar/Brokers.java → ...ub/protocol/pulsar/admin/jdk/Brokers.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 public interface Brokers {
     void healthcheck(TopicVersion topicVersion) throws PulsarAdminException;

diff --git a/...o/github/protocol/pulsar/BrokersImpl.java → ...rotocol/pulsar/admin/jdk/BrokersImpl.java b/...o/github/protocol/pulsar/BrokersImpl.java → ...rotocol/pulsar/admin/jdk/BrokersImpl.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import java.net.http.HttpResponse;
 

diff --git a/...a/io/github/protocol/pulsar/Clusters.java → ...b/protocol/pulsar/admin/jdk/Clusters.java b/...a/io/github/protocol/pulsar/Clusters.java → ...b/protocol/pulsar/admin/jdk/Clusters.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import java.util.List;
 

diff --git a/.../github/protocol/pulsar/ClustersImpl.java → ...otocol/pulsar/admin/jdk/ClustersImpl.java b/.../github/protocol/pulsar/ClustersImpl.java → ...otocol/pulsar/admin/jdk/ClustersImpl.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 

diff --git a/...github/protocol/pulsar/Configuration.java → ...tocol/pulsar/admin/jdk/Configuration.java b/...github/protocol/pulsar/Configuration.java → ...tocol/pulsar/admin/jdk/Configuration.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import lombok.Getter;
 import lombok.Setter;

diff --git a/...thub/protocol/pulsar/InnerHttpClient.java → ...col/pulsar/admin/jdk/InnerHttpClient.java b/...thub/protocol/pulsar/InnerHttpClient.java → ...col/pulsar/admin/jdk/InnerHttpClient.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 

diff --git a/...protocol/pulsar/InsecureTrustManager.java → ...ulsar/admin/jdk/InsecureTrustManager.java b/...protocol/pulsar/InsecureTrustManager.java → ...ulsar/admin/jdk/InsecureTrustManager.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import java.security.cert.CertificateException;
 import java.security.cert.X509Certificate;

diff --git a/...ithub/protocol/pulsar/JacksonService.java → ...ocol/pulsar/admin/jdk/JacksonService.java b/...ithub/protocol/pulsar/JacksonService.java → ...ocol/pulsar/admin/jdk/JacksonService.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;

diff --git a/...github/protocol/pulsar/MessageIdImpl.java → ...tocol/pulsar/admin/jdk/MessageIdImpl.java b/...github/protocol/pulsar/MessageIdImpl.java → ...tocol/pulsar/admin/jdk/MessageIdImpl.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

diff --git a/.../java/io/github/protocol/pulsar/Mode.java → ...ithub/protocol/pulsar/admin/jdk/Mode.java b/.../java/io/github/protocol/pulsar/Mode.java → ...ithub/protocol/pulsar/admin/jdk/Mode.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 public enum Mode {
     PERSISTENT,

diff --git a/...io/github/protocol/pulsar/Namespaces.java → ...protocol/pulsar/admin/jdk/Namespaces.java b/...io/github/protocol/pulsar/Namespaces.java → ...protocol/pulsar/admin/jdk/Namespaces.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import java.util.List;
 import java.util.Map;

diff --git a/...ithub/protocol/pulsar/NamespacesImpl.java → ...ocol/pulsar/admin/jdk/NamespacesImpl.java b/...ithub/protocol/pulsar/NamespacesImpl.java → ...ocol/pulsar/admin/jdk/NamespacesImpl.java
@@ -1,4 +1,4 @@
-package io.github.protocol.pulsar;
+package io.github.protocol.pulsar.admin.jdk;
 
 import com.fasterxml.jackson.core.type.TypeReference;