Merge pull request #1524 from cloudsufi/em/googlePublisher

[PLUGIN-1868] Error Management for Google Publisher

Author: Pushpender Saini

src/main/java/io/cdap/plugin/gcp/common/GCPErrorDetailsProvider.java

@@ -17,6 +17,7 @@
 package io.cdap.plugin.gcp.common;
 
 import com.google.api.client.http.HttpResponseException;
+import com.google.api.gax.rpc.ApiException;
 import com.google.common.base.Throwables;
 import io.cdap.cdap.api.exception.ErrorCategory;
 import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum;
@@ -49,10 +50,18 @@ public ProgramFailureException getExceptionDetails(Exception e, ErrorContext err
         // if causal chain already has program failure exception, return null to avoid double wrap.
         return null;
       }
+    }
+    // Reverse iterate to prioritize HttpResponseException over ApiException
+    for (int i = causalChain.size() - 1; i >= 0; i--) {
+      Throwable t = causalChain.get(i);
       if (t instanceof HttpResponseException) {
         return GCPErrorDetailsProviderUtil.getProgramFailureException((HttpResponseException) t,
           getExternalDocumentationLink(), errorContext);
       }
+      if (t instanceof ApiException) {
+        return GCPErrorDetailsProviderUtil.getProgramFailureException((ApiException) t,
+        getExternalDocumentationLink(), errorContext);
+      }
       if (t instanceof IllegalArgumentException) {
         return getProgramFailureException((IllegalArgumentException) t, errorContext);
       }

src/main/java/io/cdap/plugin/gcp/common/GCPErrorDetailsProviderUtil.java

@@ -19,6 +19,7 @@
 
 import com.google.api.client.googleapis.json.GoogleJsonResponseException;
 import com.google.api.client.http.HttpResponseException;
+import com.google.api.gax.rpc.ApiException;
 import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
 import io.cdap.cdap.api.exception.ErrorCategory;
@@ -69,21 +70,49 @@ public static ProgramFailureException getProgramFailureException(HttpResponseExc
       ErrorCodeType.HTTP, statusCode.toString(), externalDocumentationLink, e);
   }
 
+  /**
+   * Get a ProgramFailureException with the given error
+   * information from {@link ApiException}.
+   *
+   * @param e The ApiException to get the error information from.
+   * @return A ProgramFailureException with the given error information.
+   */
+  public static ProgramFailureException getProgramFailureException(ApiException e, String externalDocUrl,
+                                                                   @Nullable ErrorContext errorContext) {
+    Integer statusCode = e.getStatusCode().getCode().getHttpStatusCode();
+    ErrorUtils.ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(statusCode);
+    String errorReason = String.format("%s %s. %s. For more details, see %s", statusCode, e.getMessage(),
+     pair.getCorrectiveAction(), externalDocUrl);
+    String errorMessage = e.getMessage();
+    return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason,
+      errorContext != null ?
+        String.format(GCPErrorDetailsProvider.ERROR_MESSAGE_FORMAT, errorContext.getPhase(), e.getClass().getName(),
+          errorMessage) : String.format("%s: %s", e.getClass().getName(), errorMessage), pair.getErrorType(), true,
+      ErrorCodeType.HTTP, statusCode.toString(), externalDocUrl, e);
+  }
+
   public static ProgramFailureException getHttpResponseExceptionDetailsFromChain(Throwable e, String errorReason,
                                                                                  ErrorType errorType,
                                                                                  boolean dependency,
                                                                                  String externalDocUrl) {
     List<Throwable> causalChain = Throwables.getCausalChain(e);
+    // Check for ProgramFailureException (avoid unnecessary re-wrapping)
     for (Throwable t : causalChain) {
       if (t instanceof ProgramFailureException) {
-        // Avoid double wrap
         return (ProgramFailureException) t;
       }
+    }
+    // Reverse iterate to prioritize HttpResponseException over ApiException
+    for (int i = causalChain.size() - 1; i >= 0; i--) {
+      Throwable t = causalChain.get(i);
       if (t instanceof HttpResponseException) {
         return getProgramFailureException((HttpResponseException) t, externalDocUrl, null);
       }
+      if (t instanceof ApiException) {
+        return getProgramFailureException((ApiException) t, externalDocUrl, null);
+      }
     }
-    // If no HttpResponseException found in the causal chain, return generic program failure exception
+    // If no HttpResponseException or ApiException found in the causal chain, return generic program failure exception
     return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason,
       String.format("%s %s: %s", errorReason, e.getClass().getName(), e.getMessage()), errorType, dependency, e);
   }

src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java

@@ -81,6 +81,7 @@ public class GCPUtils {
   public static final int MILLISECONDS_MULTIPLIER = 1000;
   public static final String GCS_SUPPORTED_DOC_URL = "https://cloud.google.com/storage/docs/json_api/v1/status-codes";
   public static final String BQ_SUPPORTED_DOC_URL = "https://cloud.google.com/bigquery/docs/error-messages";
+  public static final String PUBSUB_SUPPORTED_DOC_URL = "https://cloud.google.com/pubsub/docs/reference/error-codes";
 
   /**
    * Load a service account from the local file system.

src/main/java/io/cdap/plugin/gcp/publisher/GooglePublisher.java

@@ -33,11 +33,16 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.dataset.lib.KeyValue;
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorCodeType;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
 import io.cdap.cdap.etl.api.Emitter;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchSink;
 import io.cdap.cdap.etl.api.batch.BatchSinkContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
 import io.cdap.cdap.format.StructuredRecordStringConverter;
 import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.common.batch.sink.SinkOutputFormatProvider;
@@ -116,11 +121,16 @@ public void prepareRun(BatchSinkContext context) throws IOException {
           } catch (AlreadyExistsException e1) {
             // can happen if there is a race condition. Ignore this error since all that matters is the topic exists
           } catch (ApiException e1) {
-            throw new IOException(
-              String.format("Could not auto-create topic '%s' in project '%s'. "
-                              + "Please ensure it is created before running the pipeline, "
-                              + "or ensure that the service account has permission to create the topic.",
-                            config.topic, projectId), e);
+            int statusCode = e1.getStatusCode().getCode().getHttpStatusCode();
+            String error = String.format(
+              "Could not auto-create topic '%s' in project '%s'. "
+                + "Please ensure it is created before running the pipeline, "
+                + "or ensure that the service account has permission to create the topic."
+                + "For more details, see %s",
+              config.topic, projectId, GCPUtils.PUBSUB_SUPPORTED_DOC_URL);
+            throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+            error, error, ErrorType.UNKNOWN, true, ErrorCodeType.HTTP, String.valueOf(statusCode),
+            GCPUtils.PUBSUB_SUPPORTED_DOC_URL, e1);
           }
         }
       }
@@ -130,6 +140,9 @@ public void prepareRun(BatchSinkContext context) throws IOException {
     LineageRecorder lineageRecorder = new LineageRecorder(context, config.getReferenceName());
     lineageRecorder.createExternalDataset(inputSchema);
 
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(GooglePublisherErrorDetailsProvider.class.getName()));
+
     Configuration configuration = new Configuration();
     PubSubOutputFormat.configure(configuration, config);
     context.addOutput(Output.of(config.getReferenceName(),

src/main/java/io/cdap/plugin/gcp/publisher/GooglePublisherErrorDetailsProvider.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.gcp.publisher;
+
+import io.cdap.plugin.gcp.common.GCPErrorDetailsProvider;
+import io.cdap.plugin.gcp.common.GCPUtils;
+
+/**
+ * A custom ErrorDetailsProvider for Google Publisher.
+ */
+public class GooglePublisherErrorDetailsProvider extends GCPErrorDetailsProvider {
+
+  @Override
+  protected String getExternalDocumentationLink() {
+    return GCPUtils.PUBSUB_SUPPORTED_DOC_URL;
+  }
+}

src/main/java/io/cdap/plugin/gcp/publisher/PubSubOutputFormat.java

@@ -26,10 +26,14 @@
 import com.google.pubsub.v1.ProjectTopicName;
 import com.google.pubsub.v1.PubsubMessage;
 import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.exception.ErrorCategory;
+import io.cdap.cdap.api.exception.ErrorType;
+import io.cdap.cdap.api.exception.ErrorUtils;
 import io.cdap.cdap.format.StructuredRecordStringConverter;
 import io.cdap.cdap.format.io.StructuredRecordDatumWriter;
 import io.cdap.plugin.format.avro.StructuredToAvroTransformer;
 import io.cdap.plugin.gcp.bigtable.sink.BigtableSinkConfig;
+import io.cdap.plugin.gcp.common.GCPErrorDetailsProviderUtil;
 import io.cdap.plugin.gcp.common.GCPUtils;
 import io.cdap.plugin.gcp.gcs.actions.GCSBucketCreate;
 import org.apache.avro.generic.GenericDatumWriter;
@@ -248,9 +252,11 @@ private PubsubMessage getPubSubMessage(StructuredRecord value) throws IOExceptio
       return message;
     }
 
-    private void handleErrorIfAny() throws IOException {
+    private void handleErrorIfAny() {
       if (failures.get() > errorThreshold) {
-        throw new IOException(String.format("Failed to publish %s records", failures.get()), error.get());
+        String errorMessage = String.format("Failed to publish %s records: %s.", failures.get(), error.get());
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+        errorMessage, errorMessage, ErrorType.UNKNOWN, true, null);
       }
     }
 
@@ -264,7 +270,9 @@ public void close(TaskAttemptContext taskAttemptContext) throws IOException {
           handleErrorIfAny();
         }
       } catch (ExecutionException | InterruptedException e) {
-        throw new IOException("Error publishing records to PubSub", e);
+        String errorReason = String.format("Error publishing records to PubSub: %s.", e.getMessage());
+        throw GCPErrorDetailsProviderUtil.getHttpResponseExceptionDetailsFromChain(e, errorReason, ErrorType.UNKNOWN,
+        true, GCPUtils.PUBSUB_SUPPORTED_DOC_URL);
       } finally {
         try {
           publisher.shutdown();