[CDAP-21257]Capturing Dataproc operations by 123-komal · Pull Request #16162 · cdapio/cdap

123-komal · 2026-06-11T07:47:29Z

This pull request introduces background Long-Running Operation (LRO) tracking within the Dataproc Provisioner to record the final, asynchronous outcomes of cluster creations and deletions.

Code Changes

DataprocClient.java: Implemented getLatestOperation to fetch background operation details from the regional projects.regions.operations endpoint using cluster and task-type filters.
DataprocProvisioner.java: Added a low-complexity state observer to detect terminal transitions and emit the custom provisioner.operation.response.count metric. Error payloads are mapped cleanly to ApiException wrappers.
DataprocUtils.java: Refactored the metric tag registration to export dimensions under the standard "method" key.

Testing Evidence

DataprocClientTest.java: Verify successful operations queries and clean ApiException propagation.
DataprocProvisionerTest.java: Asserted that background operation metrics are successfully captured exactly once when the cluster transitions to terminal states (CREATING -> RUNNING and DELETING -> NOT_EXISTS).
MockProvisionerContext.java: Validated that metrics are recorded and aggregated accurately without race conditions in concurrent testing scenarios.

gemini-code-assist

Code Review

This pull request introduces long-running operation (LRO) metric emission when Dataproc clusters transition to terminal states (such as RUNNING, FAILED, or NOT_EXISTS) from CREATING or DELETING. It adds a helper method to retrieve the latest operation by cluster and type, and includes corresponding unit tests. The review feedback highlights several issues: swapped arguments in a metric emission call, potential NullPointer or NoSuchElement exceptions when retrieving operations, manual gRPC status code mapping that can be simplified using standard utilities, the use of printStackTrace in production code, and leftover debug print statements in the test files.

sidhdirenge · 2026-06-11T13:57:32Z

Please create a JIRA & attach to PR.

sidhdirenge · 2026-06-11T13:58:31Z

Please add testing evidences & also change description.

vsethi09 · 2026-06-11T13:59:08Z

    this.profileName = profileName;
  }

+  private final Map<String, Integer> metricCounts = new HashMap<>();


Move variable declarations before function definitions.

sidhdirenge · 2026-06-11T13:59:16Z

Quality Gate failed

Failed conditions 53.4% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube Cloud

Add more unit tests if possible

vsethi09 · 2026-06-11T14:01:01Z

    emitMetric(context, region, metricName, imageVersion, null);
  }

+  public static void emitMetric(ProvisionerContext context, String region,


Why is this wrapper function with multiple arguments needed? It defeats the purpose of using Builder pattern for DataprocMetric.

Removed the new created emitMetric and reusing the already available method.

vsethi09 · 2026-06-11T14:02:02Z

      tags.put("imgVer", dataprocMetric.getImageVersion());
    }
+    if (!Strings.isNullOrEmpty(dataprocMetric.getOpType())) {
+      tags.put("op", dataprocMetric.getOpType());


nit: operationType or opType or operation to be clear instead of 2 character label.

In DataprocUtils.java, I mapped the tag key to "method" instead of "op" to keep it consistent.

vsethi09 · 2026-06-11T14:02:54Z

+  /**
+   * Get the latest operation by cluster name and operation type.
+   */
+  public Optional<Operation> getLatestOperationByClusterAndType(String clusterName, String operationType) {


nit: Avoid redundancy in function naming. The argument names already specify cluster name and operation type.

Noted. Updated.

vsethi09 · 2026-06-11T14:09:02Z

+        return Optional.of(operation);
+      }
+    } catch (ApiException e) {
+      LOG.warn("Failed to list operations with filter '{}': {}", filter, e.getMessage());


Is it ok to ignore the exception and return Optional.empty() response?

How would caller know if the List Operation failed vs returned empty response?

No, it is not okay to ignore the exception and return Optional.empty(). This is resolved by propagating the ApiException directly in the client method and handling it at the caller level inside the DataprocProvisioner class.

vsethi09 · 2026-06-11T14:10:05Z

+    String filter = String.format("clusterName=%s AND operationType=%s", clusterName, operationType);
+    try {
+      OperationsClient.ListOperationsPagedResponse response =
+        client.getOperationsClient().listOperations(resourceName, filter);


Should sort by option be used if the intent is to fetch the latest operation?

No, we don't need to add that because the list operation by default returns the operation based on their creation id which means latest first.

vsethi09 · 2026-06-11T14:11:31Z

+        if (op.getDone()) {
+          if (op.hasError()) {
+            int rpcCode = op.getError().getCode();
+            io.grpc.Status.Code grpcCode = io.grpc.Status.Code.INTERNAL;


Can be imported?

Similarly at other places.

vsethi09 · 2026-06-11T14:13:22Z

      @Override
-      public void count(String metricName, int delta) {}
+      public void count(String metricName, int delta) {
+        metricCounts.put(metricName, metricCounts.getOrDefault(metricName, 0) + delta);


Just curious, why is this needed?

This tracking is required so our new unit tests can capture and verify that the LRO transition metrics are being emitted correctly, the original mock context was returning empty method body that by default discard all the metrics.

sidhdirenge · 2026-06-11T14:03:26Z

    }
  }

+  private boolean isTransitionToTerminal(ClusterStatus previous, ClusterStatus current) {


maybe rename it to isEndState ?

sidhdirenge · 2026-06-11T14:05:37Z

+        }
+      }
+    } catch (Exception ex) {
+      LOG.warn("Failed to retrieve LRO operation metadata for metric emission: {}", ex.getMessage());


why not print entire ex? Any particular reason why stack trace is omitted in this log?

In the updated code, we are now correctly capturing the exception object.

sidhdirenge · 2026-06-11T14:07:32Z

+  /**
+   * Get the latest operation by cluster name and operation type.
+   */
+  public Optional<Operation> getLatestOperationByClusterAndType(String clusterName, String operationType) {


nit - getLatestOperation should be fine

Noted. Corrected.

sidhdirenge · 2026-06-11T14:08:02Z

+        return Optional.of(operation);
+      }
+    } catch (ApiException e) {
+      LOG.warn("Failed to list operations with filter '{}': {}", filter, e.getMessage());


Any particular reason why we are neglecting stack trace?

In the updated implementation, we no longer catch the exception inside DataprocClient.java. Instead, the client method now propagates the ApiException directly up to the caller, which is getLatestOperation in DataprocProvisioner, where are properly catching the stack trace.

sidhdirenge · 2026-06-11T14:12:03Z

  }

+  private boolean isTransitionToTerminal(ClusterStatus previous, ClusterStatus current) {
+    return (previous == ClusterStatus.CREATING || previous == ClusterStatus.DELETING)


can be simplified?

sidhdirenge · 2026-06-11T14:15:51Z

+      if (opOpt.isPresent()) {
+        com.google.longrunning.Operation op = opOpt.get();
+        if (op.getDone()) {
+          if (op.hasError()) {
+            int rpcCode = op.getError().getCode();
+            io.grpc.Status.Code grpcCode = io.grpc.Status.Code.INTERNAL;
+            if (rpcCode >= 0 && rpcCode < io.grpc.Status.Code.values().length) {
+              grpcCode = io.grpc.Status.Code.values()[rpcCode];
+            }
+            com.google.api.gax.rpc.StatusCode gaxStatusCode =
+              com.google.api.gax.grpc.GrpcStatusCode.of(grpcCode);
+            com.google.api.gax.rpc.ApiException apiException =
+              new com.google.api.gax.rpc.ApiException(
+                new Exception(op.getError().getMessage()), gaxStatusCode, false);
+            DataprocUtils.emitMetric(context, conf.getRegion(), metricName,
+                                     getImageVersion(context, conf), opType, new Exception(apiException));
+          } else {
+            DataprocUtils.emitMetric(context, conf.getRegion(), getImageVersion(context, conf),
+                                     metricName, opType, null);
+          }
+        }


Please simplify code, use java best practices.
There is a deep nesting (try -> if -> if -> if -> if).

rpcCode, grpcCode -> var names are confusing.

sidhdirenge · 2026-06-11T14:18:33Z

+                                     getImageVersion(context, conf), opType, new Exception(apiException));
+          } else {
+            DataprocUtils.emitMetric(context, conf.getRegion(), getImageVersion(context, conf),
+                                     metricName, opType, null);


DataprocUtils.emitMetric(context, conf.getRegion(), metricName, getImageVersion(context, conf), opType, new Exception(apiException));

DataprocUtils.emitMetric(context, conf.getRegion(), getImageVersion(context, conf), metricName, opType, null);

the parameter ordering is correct?

It was correct , we previously created a new emitMetric for case. In the updated code we are using the default emitMetric Method, so the ordering is adjusted accordingly.

sidhdirenge · 2026-06-29T11:53:36Z

+  public Optional<Operation> getLatestOperation(String clusterName, String operationType) throws ApiException {
+    String projectId = conf.getProjectId();
+    String region = conf.getRegion();
+    String resourceName = String.format("projects/%s/regions/%s/operations", projectId, region);


nit - no need to initialize separate variables in line 830, 831 -> they are just used on line 832.

sidhdirenge · 2026-06-29T12:03:27Z

+      client.getOperationsClient().listOperations(resourceName, filter);
+
+    Iterator<Operation> iterator = response.getPage().getValues().iterator();
+    return iterator.hasNext() ? Optional.of(iterator.next()) : Optional.empty();


you can directly use :

return Optional.ofNullable(Iterables.getFirst(response.getPage().getValues(), null));

instead of defining variables and re-using in next line.

sidhdirenge · 2026-06-29T12:04:27Z

+
+  @Nullable
+  private String getOpType(ClusterStatus status) {
+    if (status == ClusterStatus.CREATING) {


maybe use switch statement here?

sidhdirenge · 2026-06-29T12:27:16Z

+    return null;
+  }
+
+  private Exception createApiException(Status error) {


can be simplified :

private ApiException createApiException(Status error) { io.grpc.Status.Code grpcCode = io.grpc.Status.fromCodeValue(error.getCode()).getCode(); StatusCode gaxStatusCode = GrpcStatusCode.of(grpcCode); return new ApiException(new Exception(error.getMessage()), gaxStatusCode, false); }

it is fine to return ApiException (as per the method name)

sidhdirenge · 2026-06-29T12:28:01Z

+    }
+
+    try {
+      Optional<Operation> opOpt =


can be renamed to a better name : you can use op as well?

opOpt looks more like opOptions

renamed it as operation, not op as op name used in another variable.

vsethi09 · 2026-06-29T13:21:56Z

+  public Optional<Operation> getLatestOperation(String clusterName, String operationType) throws ApiException {
+    String projectId = conf.getProjectId();
+    String region = conf.getRegion();
+    String resourceName = String.format("projects/%s/regions/%s/operations", projectId, region);


nit: operationName?

vsethi09 · 2026-06-29T13:26:11Z

  private String profileName;
  private ErrorCategory errorCategory;

+  private final Map<String, Integer> metricCounts = new ConcurrentHashMap<>();


Move above next to other private final variables.

vsethi09 · 2026-06-29T13:26:23Z

+
+    ClusterStatus currentStatus = provisioner.getClusterStatus(localContext, cdapClusterInput);
+    Assert.assertEquals(ClusterStatus.FAILED, currentStatus);
+


nit: remove extra new line

vsethi09 · 2026-06-29T13:30:06Z

+    String resourceName = String.format("projects/%s/regions/%s/operations", projectId, region);
+    String filter = String.format("clusterName=\"%s\" AND operationType=\"%s\"", clusterName, operationType);
+
+    OperationsClient.ListOperationsPagedResponse response =


Handle ApiException using DataprocUtils.handleApiException(...)

Noted. Done.

vsethi09 · 2026-06-29T13:32:24Z

    }
  }

+  private boolean isEndState(ClusterStatus previous, ClusterStatus current) {


nit: isTerminalState for consistency,

Renamed it to hasReachedTerminalState().

vsethi09 · 2026-06-29T13:35:26Z

  }

+  private boolean isEndState(ClusterStatus previous, ClusterStatus current) {
+    return TRANSITIONING_STATES.contains(previous) && TERMINAL_STATES.contains(current);


Does this method check if the state has now transitioned to terminal state? If so rename to hasReachedTerminalState() or something better. This function is not only checking for isEndState(), so naming is confusing.

Renamed it to hasReachedTerminalState().

vsethi09 · 2026-06-29T17:00:36Z

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

+import static io.cdap.cdap.runtime.spi.common.DataprocUtils.handleApiException;


No need to import this static function. It is used as multiple places in this file as DataprocUtils.handleApiException

In general it is being done in the CDAP code base in the test classes or to import static constants / fields.

Noted. Done.

vsethi09 · 2026-06-29T17:07:31Z

+    }
+  }
+
+  private ApiException createApiException(Status error) {


Is this needed?

Refactored it now.

vsethi09 · 2026-06-29T17:09:23Z

+   * @param previousStatus the status of the cluster before the transition
+   * @param clusterName the name of the cluster
+   */
+  private void emitLroMetric(ProvisionerContext context, DataprocClient client, DataprocConf conf,


Please improve this function.

Resolving the comment, based on offline discussion.

vsethi09 · 2026-07-08T07:34:54Z

+      }
+
+      DataprocUtils.emitMetric(context, builder.build());
+    } catch (Exception ex) {


Catch specific exceptions

vsethi09 · 2026-07-08T07:37:07Z

 import java.util.function.Predicate;
 import java.util.stream.Stream;
+
+import io.cdap.cdap.runtime.spi.provisioner.RetryableProvisionException;


nit: import is misplaced.

This import is no longer needed removed.

vsethi09 · 2026-07-08T08:13:36Z

 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
+import org.powermock.api.mockito.PowerMockito;


Probably PowerMockito is not needed.

Yes, it is not needed, removed.

sidhdirenge · 2026-07-08T12:12:06Z

+    OperationsClient.ListOperationsPage page = Mockito.mock(OperationsClient.ListOperationsPage.class);
+    Mockito.when(listOperationsPagedResponse.getPage()).thenReturn(page);
+
+    Operation operation = Operation.newBuilder().setName("op-name").setDone(true).build();


should we add a test case where multiple operations are returned?

Multiple operations ? getLatestOperation just returns the latest operation, multiple operations are not returned.

Done. Added the test case.

Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated Updated

sonarqubecloud · 2026-07-08T16:46:47Z

Quality Gate passed

Issues
78 New issues
0 Accepted issues

Measures
0 Security Hotspots
80.9% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

123-komal added the build Triggers github actions build label Jun 11, 2026

gemini-code-assist Bot reviewed Jun 11, 2026

View reviewed changes

123-komal force-pushed the dataproc-operations branch 3 times, most recently from 1e9b980 to 8017a40 Compare June 11, 2026 12:32

123-komal changed the title ~~Collecting response code for Dataproc operations~~ Capturing response code for Dataproc operations Jun 11, 2026

vsethi09 reviewed Jun 11, 2026

View reviewed changes

Comment thread ...ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocClient.java Outdated

vsethi09 reviewed Jun 11, 2026

View reviewed changes

vsethi09 requested changes Jun 11, 2026

View reviewed changes

sidhdirenge reviewed Jun 11, 2026

View reviewed changes

123-komal force-pushed the dataproc-operations branch 3 times, most recently from e0dd0b5 to faec3cf Compare June 29, 2026 11:21

sidhdirenge reviewed Jun 29, 2026

View reviewed changes