feat(managedkafka): increase LRO timeout to 1 hour in client samples (#9684)

* fix: increase timeout duration in Java client

* fix: only specify modified retry settings for LRO methods

Author: rohithrajan-ai

managedkafka/examples/pom.xml

@@ -29,7 +29,7 @@
             <dependency>
                 <groupId>com.google.cloud</groupId>
                 <artifactId>libraries-bom</artifactId>
-                <version>26.40.0</version>
+                <version>26.50.0</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -40,13 +40,7 @@
     <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-managedkafka</artifactId>
-        <version>0.1.0</version>
     </dependency>
-        <dependency>
-            <groupId>com.google.api.grpc</groupId>
-            <artifactId>proto-google-cloud-managedkafka-v1</artifactId>
-            <version>0.1.0</version>
-        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

managedkafka/examples/src/main/java/examples/CreateCluster.java

@@ -17,17 +17,25 @@
 package examples;
 
 // [START managedkafka_create_cluster]
+
 import com.google.api.gax.longrunning.OperationFuture;
+import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.retrying.RetryingFuture;
+import com.google.api.gax.retrying.TimedRetryAlgorithm;
 import com.google.cloud.managedkafka.v1.AccessConfig;
 import com.google.cloud.managedkafka.v1.CapacityConfig;
 import com.google.cloud.managedkafka.v1.Cluster;
 import com.google.cloud.managedkafka.v1.CreateClusterRequest;
 import com.google.cloud.managedkafka.v1.GcpConfig;
 import com.google.cloud.managedkafka.v1.LocationName;
 import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
+import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
 import com.google.cloud.managedkafka.v1.NetworkConfig;
 import com.google.cloud.managedkafka.v1.OperationMetadata;
 import com.google.cloud.managedkafka.v1.RebalanceConfig;
+import java.time.Duration;
 import java.util.concurrent.ExecutionException;
 
 public class CreateCluster {
@@ -64,17 +72,47 @@ public static void createCluster(
             .setRebalanceConfig(rebalanceConfig)
             .build();
 
-    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
+    // Create the settings to configure the timeout for polling operations
+    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
+    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(Duration.ofHours(1L))
+            .build());
+    settingsBuilder.createClusterOperationSettings()
+        .setPollingAlgorithm(timedRetryAlgorithm);
+
+    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
+        settingsBuilder.build())) {
+
       CreateClusterRequest request =
           CreateClusterRequest.newBuilder()
               .setParent(LocationName.of(projectId, region).toString())
               .setClusterId(clusterId)
               .setCluster(cluster)
               .build();
+
       // The duration of this operation can vary considerably, typically taking between 10-40
       // minutes.
       OperationFuture<Cluster, OperationMetadata> future =
           managedKafkaClient.createClusterOperationCallable().futureCall(request);
+
+      // Get the initial LRO and print details.
+      OperationSnapshot operation = future.getInitialFuture().get();
+      System.out.printf("Cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
+          operation.getName(),
+          operation.isDone(),
+          future.getMetadata().get().toString());
+
+      while (!future.isDone()) {
+        // The pollingFuture gives us the most recent status of the operation
+        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
+        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
+        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
+            currentOp.getName(),
+            currentOp.isDone());
+      }
+
+      // NOTE: future.get() blocks completion until the operation is complete (isDone =  True)
       Cluster response = future.get();
       System.out.printf("Created cluster: %s\n", response.getName());
     } catch (ExecutionException e) {

managedkafka/examples/src/main/java/examples/DeleteCluster.java

@@ -17,14 +17,21 @@
 package examples;
 
 // [START managedkafka_delete_cluster]
+
 import com.google.api.gax.longrunning.OperationFuture;
+import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.retrying.TimedRetryAlgorithm;
 import com.google.api.gax.rpc.ApiException;
 import com.google.cloud.managedkafka.v1.ClusterName;
 import com.google.cloud.managedkafka.v1.DeleteClusterRequest;
 import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
+import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
 import com.google.cloud.managedkafka.v1.OperationMetadata;
 import com.google.protobuf.Empty;
 import java.io.IOException;
+import java.time.Duration;
 
 public class DeleteCluster {
 
@@ -38,13 +45,32 @@ public static void main(String[] args) throws Exception {
 
   public static void deleteCluster(String projectId, String region, String clusterId)
       throws Exception {
-    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
+
+    // Create the settings to configure the timeout for polling operations
+    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
+    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(Duration.ofHours(1L))
+            .build());
+    settingsBuilder.deleteClusterOperationSettings()
+        .setPollingAlgorithm(timedRetryAlgorithm);
+
+    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
+        settingsBuilder.build())) {
       DeleteClusterRequest request =
           DeleteClusterRequest.newBuilder()
               .setName(ClusterName.of(projectId, region, clusterId).toString())
               .build();
       OperationFuture<Empty, OperationMetadata> future =
           managedKafkaClient.deleteClusterOperationCallable().futureCall(request);
+
+      // Get the initial LRO and print details. CreateCluster contains sample code for polling logs.
+      OperationSnapshot operation = future.getInitialFuture().get();
+      System.out.printf("Cluster deletion started. Operation name: %s\nDone: %s\nMetadata: %s\n",
+          operation.getName(),
+          operation.isDone(),
+          future.getMetadata().get().toString());
+
       future.get();
       System.out.println("Deleted cluster");
     } catch (IOException | ApiException e) {

managedkafka/examples/src/main/java/examples/UpdateCluster.java

@@ -17,14 +17,21 @@
 package examples;
 
 // [START managedkafka_update_cluster]
+
 import com.google.api.gax.longrunning.OperationFuture;
+import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.retrying.TimedRetryAlgorithm;
 import com.google.cloud.managedkafka.v1.CapacityConfig;
 import com.google.cloud.managedkafka.v1.Cluster;
 import com.google.cloud.managedkafka.v1.ClusterName;
 import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
+import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
 import com.google.cloud.managedkafka.v1.OperationMetadata;
 import com.google.cloud.managedkafka.v1.UpdateClusterRequest;
 import com.google.protobuf.FieldMask;
+import java.time.Duration;
 import java.util.concurrent.ExecutionException;
 
 public class UpdateCluster {
@@ -48,11 +55,29 @@ public static void updateCluster(
             .build();
     FieldMask updateMask = FieldMask.newBuilder().addPaths("capacity_config.memory_bytes").build();
 
-    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create()) {
+    // Create the settings to configure the timeout for polling operations
+    ManagedKafkaSettings.Builder settingsBuilder = ManagedKafkaSettings.newBuilder();
+    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(Duration.ofHours(1L))
+            .build());
+    settingsBuilder.updateClusterOperationSettings()
+        .setPollingAlgorithm(timedRetryAlgorithm);
+
+    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
+        settingsBuilder.build())) {
       UpdateClusterRequest request =
           UpdateClusterRequest.newBuilder().setUpdateMask(updateMask).setCluster(cluster).build();
       OperationFuture<Cluster, OperationMetadata> future =
           managedKafkaClient.updateClusterOperationCallable().futureCall(request);
+
+      // Get the initial LRO and print details. CreateCluster contains sample code for polling logs.
+      OperationSnapshot operation = future.getInitialFuture().get();
+      System.out.printf("Cluster update started. Operation name: %s\nDone: %s\nMetadata: %s\n",
+          operation.getName(),
+          operation.isDone(),
+          future.getMetadata().get().toString());
+
       Cluster response = future.get();
       System.out.printf("Updated cluster: %s\n", response.getName());
     } catch (ExecutionException e) {