[fix][broker] upgrade jclouds 2.5.0 -> 2.6.0

jclouds-shaded/pom.xml

@@ -33,8 +33,17 @@
   <artifactId>jclouds-shaded</artifactId>
   <name>Apache Pulsar :: Jclouds shaded</name>
 
+  <properties>
+    <!-- for compatibility, use versions from jclouds-project
+         https://repo1.maven.org/maven2/org/apache/jclouds/jclouds-project/2.6.0/jclouds-project-2.6.0.pom -->
+    <gson.version>2.10.1</gson.version>
+    <guava.version>32.0.0-jre</guava.version>
+    <guice.version>7.0.0</guice.version>
+    <jakarta.inject.api.version>2.0.1</jakarta.inject.api.version>
+    <jakarta.ws.rs-api.version>3.0.0</jakarta.ws.rs-api.version>
+    <jakarta.annotation-api.version>2.0.0</jakarta.annotation-api.version>
+  </properties>
   <dependencies>
-
     <dependency>
       <groupId>org.apache.jclouds</groupId>
       <artifactId>jclouds-allblobstore</artifactId>
@@ -61,12 +70,48 @@
       <artifactId>jclouds-slf4j</artifactId>
       <version>${jclouds.version}</version>
     </dependency>
-    <dependency>
-      <groupId>javax.annotation</groupId>
-      <artifactId>javax.annotation-api</artifactId>
-    </dependency>
   </dependencies>
 
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>com.google.code.gson</groupId>
+        <artifactId>gson</artifactId>
+        <version>${gson.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.guava</groupId>
+        <artifactId>guava</artifactId>
+        <version>${guava.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.inject</groupId>
+        <artifactId>guice</artifactId>
+        <version>${guice.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.inject.extensions</groupId>
+        <artifactId>guice-assistedinject</artifactId>
+        <version>${guice.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>jakarta.inject</groupId>
+        <artifactId>jakarta.inject-api</artifactId>
+        <version>${jakarta.inject.api.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>jakarta.ws.rs</groupId>
+        <artifactId>jakarta.ws.rs-api</artifactId>
+        <version>${jakarta.ws.rs-api.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>jakarta.annotation</groupId>
+        <artifactId>jakarta.annotation-api</artifactId>
+        <version>${jakarta.annotation-api.version}</version>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
   <build>
     <plugins>
       <plugin>
@@ -97,13 +142,13 @@
                   <include>com.google.inject.extensions:guice-multibindings</include>
                   <include>com.google.code.gson:gson</include>
                   <include>org.apache.httpcomponents:*</include>
-                  <include>javax.ws.rs:*</include>
                   <include>com.jamesmurty.utils:*</include>
                   <include>net.iharder:*</include>
                   <include>aopalliance:*</include>
-                  <include>javax.inject:*</include>
-                  <include>javax.annotation:*</include>
                   <include>com.google.errorprone:*</include>
+                  <include>jakarta.inject:jakarta.inject-api</include>
+                  <include>jakarta.annotation:jakarta.annotation-api</include>
+                  <include>jakarta.ws.rs:jakarta.ws.rs-api</include>
                 </includes>
               </artifactSet>
 
@@ -112,10 +157,6 @@
                   <pattern>com.google</pattern>
                   <shadedPattern>org.apache.pulsar.jcloud.shade.com.google</shadedPattern>
                 </relocation>
-                <relocation>
-                  <pattern>javax.ws</pattern>
-                  <shadedPattern>org.apache.pulsar.jcloud.shade.javax.ws</shadedPattern>
-                </relocation>
                 <relocation>
                   <pattern>com.jamesmurty.utils</pattern>
                   <shadedPattern>org.apache.pulsar.jcloud.shade.com.jamesmurty.utils</shadedPattern>
@@ -129,18 +170,17 @@
                   <shadedPattern>org.apache.pulsar.jcloud.shade.net.iharder</shadedPattern>
                 </relocation>
                 <relocation>
-                  <pattern>javax.inject</pattern>
-                  <shadedPattern>org.apache.pulsar.jcloud.shade.javax.inject</shadedPattern>
+                  <pattern>com.google.errorprone</pattern>
+                  <shadedPattern>org.apache.pulsar.jcloud.shade.com.google.errorprone</shadedPattern>
                 </relocation>
                 <relocation>
-                  <pattern>javax.annotation</pattern>
-                  <shadedPattern>org.apache.pulsar.jcloud.shade.javax.annotation</shadedPattern>
+                  <pattern>jakarta</pattern>
+                  <shadedPattern>org.apache.pulsar.jcloud.shade.jakarta</shadedPattern>
                 </relocation>
                 <relocation>
-                  <pattern>com.google.errorprone</pattern>
-                  <shadedPattern>org.apache.pulsar.jcloud.shade.com.google.errorprone</shadedPattern>
+                  <pattern>org.aopalliance</pattern>
+                  <shadedPattern>org.apache.pulsar.jcloud.shade.org.aopalliance</shadedPattern>
                 </relocation>
-
               </relocations>
               <transformers>
                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>

pip/pip-343.md

@@ -0,0 +1,143 @@
+# PIP-343: Use picocli instead of jcommander
+
+# Motivation
+
+We use the [jcommander](https://github.com/cbeust/jcommander) to build the CLI tool, which is a good library, and is
+stable, but it misses modern CLI features likes autocompletion, flag/command suggestion, native image, etc.
+
+These features are very important because there are many commands in the CLI, but the jcommander doesn't give friendly
+hints when we use incorrect flags/commands, which makes the user experience not very friendly.
+
+In modern times, the [picocli](https://github.com/remkop/picocli) supports these features, which is a popular library.
+
+The following is some comparison between jcommander and picocli:
+
+- Error prompt:
+  ```
+  bin/pulsar-admin clusters update cluster-a -b
+
+  # jcommander
+  Need to provide just 1 parameter
+
+  # picocli
+  Unknown option: '-b'
+  ```
+
+- Command suggestion:
+  ```
+  bin/pulsar-admin cluste
+
+  # jcommander
+  Expected a command, got cluste
+
+  # picocli
+  Unmatched argument at index 0: 'cluste'
+  Did you mean: pulsar-admin clusters?
+  ```
+
+# Goals
+
+## In Scope
+
+Use the picocli instead of the jcommander in our CLI tool:
+
+- bin/pulsar
+- bin/pulsar-admin
+- bin/pulsar-client
+- bin/pulsar-shell
+- bin/pulsar-perf
+
+I'm sure this will greatly improve the user experience, and in the future we can also consider using native images to
+reduce runtime, and improve the CLI document based on picocli.
+
+## Out Scope
+
+This PR simply replaces jcommander and does not introduce any enhancements.
+
+In the CLI, [autocomplete](https://picocli.info/autocomplete.html) is an important feature, and after this PIP is
+complete I will make a new PIP to support this feature.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The jcommander and picocli have similar APIs, this will make the migration task very simple.
+
+This is [utility argument syntax](https://pubs.opengroup.org/onlinepubs/9699919799/basedefs/V1_chap12.html):
+
+```
+utility_name[-a][-b][-c option_argument]
+    [-d|-e][-f[option_argument]][operand...]
+```
+
+1. Use `@Command` instead of `@Parameters` to define the class as a command:
+
+  ```java
+  @Command(name = "my-command", description = "Operations on persistent topics")
+  public class MyCommand {
+
+  }
+  ```
+
+2. Use `@Option` instead of `@Parameter` to defined the option of command:
+
+  ```java
+  @Option(names = {"-r", "--role"})
+  private String role;
+  ```
+
+3. Use `@Parameters` to get the operand of command:
+
+  ```java
+  @Parameters(description = "persistent://tenant/namespace/topic", arity = "1")
+  private String topicName;
+  ```
+
+4. Migrate jcommander converter to picocli converter:
+
+  ```java
+  public class TimeUnitToMillisConverter implements ITypeConverter<Long> {
+    @Override
+    public Long convert(String value) throws Exception {
+      return TimeUnit.SECONDS.toMillis(RelativeTimeUtil.parseRelativeTimeInSeconds(value));
+    }
+  }
+  ```
+
+5. Add the picocli entrypoint:
+
+  ```java
+  @Command
+  public class MyCommand implements Callable<Integer> {
+    // Picocli entrypoint.
+    @Override
+    public Integer call() throws Exception {
+        // TODO
+        // run(); 
+        return 0;
+    }
+  }
+  ``` 
+
+The above is a common migration approach, and then we need to consider pulsar-shell and custom command separately.
+
+- pulsar-shell
+
+  This is an interactive shell based on jline3 and jcommander, which includes pulsar-admin and pulsar-client commands.
+  The jcommander does not provide autocompletion because we have implemented it ourselves. In picocli, they
+  have [picocli-shell-jline3](https://github.com/remkop/picocli/blob/main/picocli-shell-jline3) to help us quickly build
+  the interactive shell.
+
+- custom command:
+
+  This is an extension of pulsar-admin, and the plugin's implementation does not depend on jcommander. Since the bridge
+  is used, we only need to change the generator code based on picocli.
+
+# Backward & Forward Compatibility
+
+Fully compatible.
+
+# Links
+
+* Mailing List discussion thread: https://lists.apache.org/thread/ydg1q064cd11pxwz693frtk4by74q32f
+* Mailing List voting thread: https://lists.apache.org/thread/1bpsr6tkgm00bb66dt2s74r15o4b37s3

pom.xml

@@ -185,7 +185,7 @@ flexible messaging model and an intuitive client API.</description>
     <aws-sdk.version>1.12.262</aws-sdk.version>
     <avro.version>1.11.3</avro.version>
     <joda.version>2.10.10</joda.version>
-    <jclouds.version>2.5.0</jclouds.version>
+    <jclouds.version>2.6.0</jclouds.version>
     <guice.version>5.1.0</guice.version>
     <sqlite-jdbc.version>3.42.0.0</sqlite-jdbc.version>
     <mysql-jdbc.version>8.0.11</mysql-jdbc.version>

pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java

@@ -32,8 +32,14 @@ public interface RawReader {
      */
 
     static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
+        return create(client, topic, subscription, true);
+    }
+
+    static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription,
+                                               boolean createTopicIfDoesNotExist) {
         CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
-        RawReader r = new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future);
+        RawReader r =
+                new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future, createTopicIfDoesNotExist);
         return future.thenApply(__ -> r);
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java

@@ -51,7 +51,8 @@ public class RawReaderImpl implements RawReader {
     private RawConsumerImpl consumer;
 
     public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
-                         CompletableFuture<Consumer<byte[]>> consumerFuture) {
+                         CompletableFuture<Consumer<byte[]>> consumerFuture,
+                         boolean createTopicIfDoesNotExist) {
         consumerConfiguration = new ConsumerConfigurationData<>();
         consumerConfiguration.getTopicNames().add(topic);
         consumerConfiguration.setSubscriptionName(subscription);
@@ -61,8 +62,7 @@ public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
         consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
         consumerConfiguration.setAckReceiptEnabled(true);
 
-        consumer = new RawConsumerImpl(client, consumerConfiguration,
-                                       consumerFuture);
+        consumer = new RawConsumerImpl(client, consumerConfiguration, consumerFuture, createTopicIfDoesNotExist);
     }
 
     @Override
@@ -111,7 +111,7 @@ static class RawConsumerImpl extends ConsumerImpl<byte[]> {
         final Queue<CompletableFuture<RawMessage>> pendingRawReceives;
 
         RawConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<byte[]> conf,
-                CompletableFuture<Consumer<byte[]>> consumerFuture) {
+                CompletableFuture<Consumer<byte[]>> consumerFuture, boolean createTopicIfDoesNotExist) {
             super(client,
                     conf.getSingleTopic(),
                     conf,
@@ -123,7 +123,7 @@ static class RawConsumerImpl extends ConsumerImpl<byte[]> {
                     MessageId.earliest,
                     0 /* startMessageRollbackDurationInSec */,
                     Schema.BYTES, null,
-                    false
+                    createTopicIfDoesNotExist
             );
             incomingRawMessages = new GrowableArrayBlockingQueue<>();
             pendingRawReceives = new ConcurrentLinkedQueue<>();

pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java

@@ -56,7 +56,7 @@ public Compactor(ServiceConfiguration conf,
     }
 
     public CompletableFuture<Long> compact(String topic) {
-        return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION).thenComposeAsync(
+        return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION, false).thenComposeAsync(
                 this::compactAndCloseReader, scheduler);
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -36,15 +37,18 @@
 import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.RawMessage;
 import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
 import org.apache.pulsar.common.api.proto.MessageMetadata;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.policies.data.TopicStats;
 import org.apache.pulsar.common.protocol.Commands;
 import org.awaitility.Awaitility;
 import org.testng.Assert;
@@ -496,4 +500,23 @@ public void testReadCancellationOnClose() throws Exception {
             }
         }
     }
+
+    @Test
+    public void testAutoCreateTopic() throws ExecutionException, InterruptedException, PulsarAdminException {
+        String topic = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
+
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
+        TopicStats stats = admin.topics().getStats(topic);
+        Assert.assertNotNull(stats);
+        reader.closeAsync().join();
+
+        String topic2 = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
+        try {
+            reader = RawReader.create(pulsarClient, topic2, subscription, false).get();
+            Assert.fail();
+        } catch (Exception e) {
+            Assert.assertTrue(e.getCause() instanceof PulsarClientException.TopicDoesNotExistException);
+        }
+        reader.closeAsync().join();
+    }
 }