[fix apache#9640] remove pulsar-client-admin-api dependency : `puls…

…ar-client-original` (apache#10563) * remove org.apache.pulsar.client.impl.conf.ClientConfigurationData * use MessageId instead of MessageIdImpl * get getClientConfigData back * fix io-kafka dependency * fix java-test-functions dependency * add pulsar-client-original to pulsar-functions-runtime-all * fix offload tests * fix import * change Gson to Jackson for OffloadProcessStatus * move OffloadProcessStatus as interface, add InterfaceDefaultMapperModule for interface resolver
AlphaWang · May 17, 2021 · 7b9c846 · 7b9c846
1 parent ed935a7
commit 7b9c846
Show file tree

Hide file tree

Showing 13 changed files with 139 additions and 47 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java
@@ -28,12 +28,13 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
 import com.google.common.collect.Sets;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.mledger.LedgerOffloader;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -104,12 +105,12 @@ private void testOffload(String topicName, String mlName) throws Exception {
         ManagedLedgerInfo info = pulsar.getManagedLedgerFactory().getManagedLedgerInfo(mlName);
         assertEquals(info.ledgers.size(), 2);
 
-        assertEquals(admin.topics().offloadStatus(topicName).status,
+        assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
                             LongRunningProcessStatus.Status.NOT_RUN);
 
         admin.topics().triggerOffload(topicName, currentId);
 
-        assertEquals(admin.topics().offloadStatus(topicName).status,
+        assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
                             LongRunningProcessStatus.Status.RUNNING);
 
         try {
@@ -122,9 +123,9 @@ private void testOffload(String topicName, String mlName) throws Exception {
         // fail first time
         promise.completeExceptionally(new Exception("Some random failure"));
 
-        assertEquals(admin.topics().offloadStatus(topicName).status,
+        assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
                             LongRunningProcessStatus.Status.ERROR);
-        Assert.assertTrue(admin.topics().offloadStatus(topicName).lastError.contains("Some random failure"));
+        Assert.assertTrue(admin.topics().offloadStatus(topicName).getLastError().contains("Some random failure"));
 
         // Try again
         doReturn(CompletableFuture.completedFuture(null))
@@ -133,12 +134,14 @@ private void testOffload(String topicName, String mlName) throws Exception {
         admin.topics().triggerOffload(topicName, currentId);
 
         Awaitility.await().untilAsserted(() ->
-                assertEquals(admin.topics().offloadStatus(topicName).status,
+                assertEquals(admin.topics().offloadStatus(topicName).getStatus(),
                 LongRunningProcessStatus.Status.SUCCESS));
-        MessageIdImpl firstUnoffloaded = admin.topics().offloadStatus(topicName).firstUnoffloadedMessage;
+        MessageId firstUnoffloaded = admin.topics().offloadStatus(topicName).getFirstUnoffloadedMessage();
+        assertTrue(firstUnoffloaded instanceof MessageIdImpl);
+        MessageIdImpl firstUnoffloadedMessage = (MessageIdImpl) firstUnoffloaded;
         // First unoffloaded is the first entry of current ledger
-        assertEquals(firstUnoffloaded.getLedgerId(), info.ledgers.get(1).ledgerId);
-        assertEquals(firstUnoffloaded.getEntryId(), 0);
+        assertEquals(firstUnoffloadedMessage.getLedgerId(), info.ledgers.get(1).ledgerId);
+        assertEquals(firstUnoffloadedMessage.getEntryId(), 0);
 
         verify(offloader, times(2)).offload(any(), any(), any());
     }

diff --git a/pulsar-client-admin-api/pom.xml b/pulsar-client-admin-api/pom.xml
@@ -33,16 +33,16 @@
     <name>Pulsar Client Admin :: API</name>
 
     <dependencies>
+
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>pulsar-common</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
 
         <dependency>
-            <groupId>${project.groupId}</groupId>
-            <artifactId>pulsar-client-original</artifactId>
-            <version>${project.parent.version}</version>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
         </dependency>
 
     </dependencies>

diff --git a/...r-client-admin-api/src/main/java/org/apache/pulsar/client/admin/OffloadProcessStatus.java b/...r-client-admin-api/src/main/java/org/apache/pulsar/client/admin/OffloadProcessStatus.java
@@ -18,38 +18,30 @@
  */
 package org.apache.pulsar.client.admin;
 
+import org.apache.pulsar.client.admin.utils.DefaultImplementation;
 import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.impl.MessageIdImpl;
 
 /**
- * Status of offload process.
+ * interface class of Status of offload process.
  */
-public class OffloadProcessStatus extends LongRunningProcessStatus {
+public interface OffloadProcessStatus {
 
-    public MessageIdImpl firstUnoffloadedMessage;
+    MessageId getFirstUnoffloadedMessage();
+    String getLastError();
+    LongRunningProcessStatus.Status getStatus();
 
-    public OffloadProcessStatus() {
-        super(Status.NOT_RUN, "");
-        firstUnoffloadedMessage = (MessageIdImpl) MessageId.earliest;
+    static OffloadProcessStatus forStatus(LongRunningProcessStatus.Status status) {
+        return DefaultImplementation.newOffloadProcessStatus(status, "", MessageId.earliest);
     }
 
-    private OffloadProcessStatus(Status status, String lastError,
-                                 MessageIdImpl firstUnoffloadedMessage) {
-        this.status = status;
-        this.lastError = lastError;
-        this.firstUnoffloadedMessage = firstUnoffloadedMessage;
+    static OffloadProcessStatus forError(String lastError) {
+        return DefaultImplementation.newOffloadProcessStatus(LongRunningProcessStatus.Status.ERROR, lastError,
+                MessageId.earliest);
     }
 
-    public static OffloadProcessStatus forStatus(Status status) {
-        return new OffloadProcessStatus(status, "", (MessageIdImpl) MessageId.earliest);
+    static OffloadProcessStatus forSuccess(MessageId messageId) {
+        return DefaultImplementation.newOffloadProcessStatus(LongRunningProcessStatus.Status.SUCCESS, "",
+                messageId);
     }
 
-    public static OffloadProcessStatus forError(String lastError) {
-        return new OffloadProcessStatus(Status.ERROR, lastError,
-                (MessageIdImpl) MessageId.earliest);
-    }
-
-    public static OffloadProcessStatus forSuccess(MessageIdImpl messageId) {
-        return new OffloadProcessStatus(Status.SUCCESS, "", messageId);
-    }
 }
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
@@ -20,7 +20,6 @@
 
 import java.io.Closeable;
 import org.apache.pulsar.client.admin.utils.DefaultImplementation;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
 
@@ -147,11 +146,6 @@ static PulsarAdminBuilder builder() {
      */
     String getServiceUrl();
 
-    /**
-     * @return the client Configuration Data that is being used
-     */
-    ClientConfigurationData getClientConfigData();
-
     /**
      * @return the schemas
      */

diff --git a/...t-admin-api/src/main/java/org/apache/pulsar/client/admin/utils/DefaultImplementation.java b/...t-admin-api/src/main/java/org/apache/pulsar/client/admin/utils/DefaultImplementation.java
@@ -18,8 +18,12 @@
  */
 package org.apache.pulsar.client.admin.utils;
 
+import java.lang.reflect.Constructor;
 import lombok.experimental.UtilityClass;
+import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+import org.apache.pulsar.client.admin.OffloadProcessStatus;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.api.MessageId;
 
 /**
  * Helper class for class instantiations and it also contains methods to work with schemas.
@@ -30,7 +34,18 @@ public class DefaultImplementation {
     private static final Class<PulsarAdminBuilder> ADMIN_CLIENT_BUILDER_IMPL = ReflectionUtils.newClassInstance(
             "org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl");
 
+    private static final Constructor<OffloadProcessStatus> OFFLOAD_PROCESS_STATUS_IMPL_status_string_messageid =
+            ReflectionUtils.getConstructor(
+            "org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl",
+                    LongRunningProcessStatus.Status.class, String.class, MessageId.class);
+
     public static PulsarAdminBuilder newAdminClientBuilder() {
         return ReflectionUtils.catchExceptions(() -> ADMIN_CLIENT_BUILDER_IMPL.newInstance());
     }
+
+    public static OffloadProcessStatus newOffloadProcessStatus(LongRunningProcessStatus.Status status, String lastError
+            , MessageId messageId) {
+        return ReflectionUtils.catchExceptions(() -> OFFLOAD_PROCESS_STATUS_IMPL_status_string_messageid.newInstance(
+                status, lastError, messageId));
+    }
 }
diff --git a/...ient-admin/src/main/java/org/apache/pulsar/client/admin/internal/JacksonConfigurator.java b/...ient-admin/src/main/java/org/apache/pulsar/client/admin/internal/JacksonConfigurator.java
@@ -19,8 +19,11 @@
 package org.apache.pulsar.client.admin.internal;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.module.SimpleAbstractTypeResolver;
+import com.fasterxml.jackson.databind.module.SimpleModule;
 import javax.ws.rs.ext.ContextResolver;
 import javax.ws.rs.ext.Provider;
+import org.apache.pulsar.client.admin.OffloadProcessStatus;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 
 /**
@@ -33,6 +36,20 @@ public class JacksonConfigurator implements ContextResolver<ObjectMapper> {
 
     public JacksonConfigurator() {
         mapper = ObjectMapperFactory.create();
+        setInterfaceDefaultMapperModule();
+    }
+
+    private void setInterfaceDefaultMapperModule() {
+        SimpleModule module = new SimpleModule("InterfaceDefaultMapperModule");
+
+        // we not specific @JsonDeserialize annotation in OffloadProcessStatus
+        // because we do not want to have jackson dependency in pulsar-client-admin-api
+        // In this case we use SimpleAbstractTypeResolver to map interfaces to impls
+        SimpleAbstractTypeResolver resolver = new SimpleAbstractTypeResolver();
+        resolver.addMapping(OffloadProcessStatus.class, OffloadProcessStatusImpl.class);
+
+        module.setAbstractTypes(resolver);
+        mapper.registerModule(module);
     }
 
     @Override

diff --git a/...admin/src/main/java/org/apache/pulsar/client/admin/internal/OffloadProcessStatusImpl.java b/...admin/src/main/java/org/apache/pulsar/client/admin/internal/OffloadProcessStatusImpl.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.admin.internal;
+
+import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+import org.apache.pulsar.client.admin.OffloadProcessStatus;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+
+/**
+ * Status of offload process.
+ */
+public class OffloadProcessStatusImpl extends LongRunningProcessStatus implements OffloadProcessStatus {
+    public MessageIdImpl firstUnoffloadedMessage;
+
+    public OffloadProcessStatusImpl() {
+        status = Status.NOT_RUN;
+        lastError = "";
+        firstUnoffloadedMessage = (MessageIdImpl) MessageId.earliest;
+    }
+
+    public OffloadProcessStatusImpl(Status status, String lastError,
+                                 MessageId firstUnoffloadedMessage) {
+        this.status = status;
+        this.lastError = lastError;
+        this.firstUnoffloadedMessage = (MessageIdImpl) firstUnoffloadedMessage;
+    }
+
+    @Override
+    public MessageId getFirstUnoffloadedMessage() {
+        return firstUnoffloadedMessage;
+    }
+
+    @Override
+    public String getLastError() {
+        return lastError;
+    }
+
+    @Override
+    public Status getStatus() {
+        return status;
+    }
+}
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
@@ -116,7 +116,6 @@ public void setup() throws Exception {
         this.functions = mock(Functions.class);
         when(admin.functions()).thenReturn(functions);
         when(admin.getServiceUrl()).thenReturn("http://localhost:1234");
-        when(admin.getClientConfigData()).thenReturn(new ClientConfigurationData());
         this.cmd = new CmdFunctions(() -> admin);
         this.cmdSinks = new CmdSinks(() -> admin);
         this.cmdSources = new CmdSources(() -> admin);

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
@@ -54,6 +54,7 @@
 import org.apache.pulsar.client.admin.Schemas;
 import org.apache.pulsar.client.admin.Tenants;
 import org.apache.pulsar.client.admin.Topics;
+import org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl;
 import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -1121,7 +1122,7 @@ public boolean matches(Long timestamp) {
         cmdTopics.run(split("offload persistent://myprop/clust/ns1/ds1 -s 1k"));
         verify(mockTopics).triggerOffload("persistent://myprop/clust/ns1/ds1", new MessageIdImpl(2, 0, -1));
 
-        when(mockTopics.offloadStatus("persistent://myprop/clust/ns1/ds1")).thenReturn(new OffloadProcessStatus());
+        when(mockTopics.offloadStatus("persistent://myprop/clust/ns1/ds1")).thenReturn(new OffloadProcessStatusImpl());
         cmdTopics.run(split("offload-status persistent://myprop/clust/ns1/ds1"));
         verify(mockTopics).offloadStatus("persistent://myprop/clust/ns1/ds1");
 

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
@@ -44,6 +44,7 @@
 import java.util.stream.Collectors;
 
 import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+import org.apache.pulsar.client.admin.OffloadProcessStatus;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.Topics;
@@ -1035,13 +1036,13 @@ void run() throws PulsarAdminException {
             String persistentTopic = validatePersistentTopic(params);
 
             try {
-                LongRunningProcessStatus status = getTopics().offloadStatus(persistentTopic);
-                while (wait && status.status == LongRunningProcessStatus.Status.RUNNING) {
+                OffloadProcessStatus status = getTopics().offloadStatus(persistentTopic);
+                while (wait && status.getStatus() == LongRunningProcessStatus.Status.RUNNING) {
                     Thread.sleep(1000);
                     status = getTopics().offloadStatus(persistentTopic);
                 }
 
-                switch (status.status) {
+                switch (status.getStatus()) {
                 case NOT_RUN:
                     System.out.println("Offload has not been run for " + persistentTopic
                                        + " since broker startup");
@@ -1054,7 +1055,7 @@ void run() throws PulsarAdminException {
                     break;
                 case ERROR:
                     System.out.println("Error in offload");
-                    throw new PulsarAdminException("Error offloading: " + status.lastError);
+                    throw new PulsarAdminException("Error offloading: " + status.getLastError());
                 }
             } catch (InterruptedException e) {
                 throw new PulsarAdminException(e);

diff --git a/pulsar-functions/runtime-all/pom.xml b/pulsar-functions/runtime-all/pom.xml
@@ -48,7 +48,7 @@
 
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>pulsar-client-api</artifactId>
+      <artifactId>pulsar-client-original</artifactId>
       <version>${project.version}</version>
     </dependency>
 

diff --git a/pulsar-io/kafka/pom.xml b/pulsar-io/kafka/pom.xml
@@ -39,6 +39,12 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-client-original</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>

diff --git a/tests/docker-images/java-test-functions/pom.xml b/tests/docker-images/java-test-functions/pom.xml
@@ -34,6 +34,11 @@
             <artifactId>pulsar-io-core</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client-original</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
     <packaging>jar</packaging>