[ISSUE #1133] The java client adds the maxStartupAttempts parameter to configure the retry times at startup (#1134)

Author: qianye1001

.github/workflows/java_build.yml

@@ -28,11 +28,11 @@ jobs:
     steps:
       - name: Checkout Current Repository
         uses: actions/checkout@v3
-      # Use JDK 17.
-      - name: Use JDK 17
+      # Use JDK 21.
+      - name: Use JDK 21
         uses: actions/setup-java@v3
         with:
-          java-version: 17
+          java-version: 21
           distribution: "adopt"
       # Build the code of the current repository, skipping tests and code style checks.
       - name: Build Current Repository
@@ -66,10 +66,10 @@ jobs:
         run: |
           sed -i 's/org\.apache\.rocketmq:rocketmq-client-java:[^"]*/org.apache.rocketmq:rocketmq-client-java:${{ steps.get_version.outputs.version }}/' instrumentation/rocketmq/rocketmq-client/rocketmq-client-5.0/testing/build.gradle.kts
       # Use JDK 17.
-      - name: Use JDK 17
+      - name: Use JDK 21
         uses: actions/setup-java@v3
         with:
-          java-version: 17
+          java-version: 21
           distribution: "adopt"
           cache: gradle
       # Build the rocketmq opentelemetry test.

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfiguration.java

@@ -29,18 +29,20 @@ public class ClientConfiguration {
     private final Duration requestTimeout;
     private final boolean sslEnabled;
     private final String namespace;
+    private final int maxStartupAttempts;
 
     /**
      * The caller is supposed to have validated the arguments and handled throwing exceptions or
      * logging warnings already, so we avoid repeating args check here.
      */
     ClientConfiguration(String endpoints, SessionCredentialsProvider sessionCredentialsProvider,
-        Duration requestTimeout, boolean sslEnabled, String namespace) {
+        Duration requestTimeout, boolean sslEnabled, String namespace, int maxStartupAttempts) {
         this.endpoints = endpoints;
         this.sessionCredentialsProvider = sessionCredentialsProvider;
         this.requestTimeout = requestTimeout;
         this.sslEnabled = sslEnabled;
         this.namespace = namespace;
+        this.maxStartupAttempts = maxStartupAttempts;
     }
 
     public static ClientConfigurationBuilder newBuilder() {
@@ -66,4 +68,8 @@ public boolean isSslEnabled() {
     public String getNamespace() {
         return namespace;
     }
+
+    public int getMaxStartupAttempts() {
+        return maxStartupAttempts;
+    }
 }

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientConfigurationBuilder.java

@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.client.apis;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.time.Duration;
@@ -32,6 +33,7 @@ public class ClientConfigurationBuilder {
     private Duration requestTimeout = Duration.ofSeconds(3);
     private boolean sslEnabled = true;
     private String namespace = "";
+    private int maxStartupAttempts = 3;
 
     /**
      * Configure the access point with which the SDK should communicate.
@@ -93,6 +95,18 @@ public ClientConfigurationBuilder setNamespace(String namespace) {
         return this;
     }
 
+    /**
+     * Configure maxStartupAttempts for client
+     *
+     * @param maxStartupAttempts max attempt times when client startup
+     * @return The {@link ClientConfigurationBuilder} instance, to allow for method chaining.
+     */
+    public ClientConfigurationBuilder setMaxStartupAttempts(int maxStartupAttempts) {
+        checkArgument(maxStartupAttempts > 0, "maxStartupAttempts should more than 0");
+        this.maxStartupAttempts = maxStartupAttempts;
+        return this;
+    }
+
     /**
      * Finalize the build of {@link ClientConfiguration}.
      *
@@ -101,6 +115,7 @@ public ClientConfigurationBuilder setNamespace(String namespace) {
     public ClientConfiguration build() {
         checkNotNull(endpoints, "endpoints should not be null");
         checkNotNull(requestTimeout, "requestTimeout should not be null");
-        return new ClientConfiguration(endpoints, sessionCredentialsProvider, requestTimeout, sslEnabled, namespace);
+        return new ClientConfiguration(endpoints, sessionCredentialsProvider, requestTimeout, sslEnabled, namespace,
+            maxStartupAttempts);
     }
 }

java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerSingleton.java

@@ -54,6 +54,9 @@ private static Producer buildProducer(TransactionChecker checker, String... topi
             // On some Windows platforms, you may encounter SSL compatibility issues. Try turning off the SSL option in
             // client configuration to solve the problem please if SSL is not essential.
             // .enableSsl(false)
+            // Due to the lazy loading of gRPC, when the network conditions are poor or the load of the application
+            // at startup is high, the first startup may fail, and you can try multiple startups.
+            // .setMaxStartupAttempts(3)
             .setCredentialProvider(sessionCredentialsProvider)
             .build();
         final ProducerBuilder builder = provider.newProducerBuilder()

java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java

@@ -186,12 +186,24 @@ protected void startUp() throws Exception {
         // Fetch topic route from remote.
         log.info("Begin to fetch topic(s) route data from remote during client startup, clientId={}, topics={}",
             clientId, topics);
-        for (String topic : topics) {
-            final ListenableFuture<TopicRouteData> future = fetchTopicRoute(topic);
-            future.get();
+        for (int attempt = 1; attempt <= clientConfiguration.getMaxStartupAttempts(); attempt++) {
+            try {
+                for (String topic : topics) {
+                    final ListenableFuture<TopicRouteData> future = fetchTopicRoute(topic);
+                    future.get();
+                }
+                log.info("Fetch topic route data from remote successfully during startup, clientId={}, topics={}",
+                    clientId, topics);
+                break;
+            } catch (Exception e) {
+                log.error("Fetch topics failed when client start, clientId={}, topics={}, attemptTime={}", clientId,
+                    topics, attempt, e);
+                if (attempt == clientConfiguration.getMaxStartupAttempts()) {
+                    throw new RuntimeException(
+                        String.format("Failed to fetch topics after %d attempts", attempt), e);
+                }
+            }
         }
-        log.info("Fetch topic route data from remote successfully during startup, clientId={}, topics={}",
-            clientId, topics);
         // Update route cache periodically.
         final ScheduledExecutorService scheduler = clientManager.getScheduler();
         this.updateRouteCacheFuture = scheduler.scheduleWithFixedDelay(() -> {