Error Management Big-Table

Author: psainics

src/main/java/io/cdap/plugin/gcp/bigtable/common/BigtableErrorDetailsProvider.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.bigtable.common;
+
+import io.cdap.plugin.gcp.common.GCPErrorDetailsProvider;
+import io.cdap.plugin.gcp.common.GCPUtils;
+
+/**
+ * A custom ErrorDetailsProvider for BigTable plugins.
+ */
+public class BigtableErrorDetailsProvider extends GCPErrorDetailsProvider {
+
+  @Override
+  protected String getExternalDocumentationLink() {
+    return GCPUtils.BIG_TABLE_SUPPORTED_DOC_URL;
+  }
+}

src/main/java/io/cdap/plugin/gcp/bigtable/sink/BigtableSink.java

@@ -34,10 +34,13 @@
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 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.plugin.common.ConfigUtil;
 import io.cdap.plugin.common.LineageRecorder;
+import io.cdap.plugin.gcp.bigtable.common.BigtableErrorDetailsProvider;
 import io.cdap.plugin.gcp.bigtable.common.HBaseColumn;
 import io.cdap.plugin.gcp.common.SourceOutputFormatProvider;
+import io.cdap.plugin.gcp.gcs.GCSErrorDetailsProvider;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HTableDescriptor;
@@ -139,6 +142,8 @@ public void prepareRun(BatchSinkContext context) {
     // Both emitLineage and setOutputFormat internally try to create an external dataset if it does not already exists.
     // We call emitLineage before since it creates the dataset with schema.
     emitLineage(context);
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(BigtableErrorDetailsProvider.class.getName()));
     context.addOutput(Output.of(config.getReferenceName(),
                                 new SourceOutputFormatProvider(BigtableOutputFormat.class, conf)));
   }

src/main/java/io/cdap/plugin/gcp/bigtable/source/BigtableSource.java

@@ -25,16 +25,21 @@
 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.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.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 import io.cdap.cdap.etl.api.batch.BatchSource;
 import io.cdap.cdap.etl.api.batch.BatchSourceContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
 import io.cdap.plugin.common.ConfigUtil;
 import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.common.SourceInputFormatProvider;
+import io.cdap.plugin.gcp.bigtable.common.BigtableErrorDetailsProvider;
 import io.cdap.plugin.gcp.bigtable.common.HBaseColumn;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -121,6 +126,8 @@ public void prepareRun(BatchSourceContext context) {
     // Both emitLineage and setOutputFormat internally try to create an external dataset if it does not already exists.
     // We call emitLineage before since it creates the dataset with schema.
     emitLineage(context, configuredSchema);
+    // set error details provider
+    context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(BigtableErrorDetailsProvider.class.getName()));
     context.setInput(Input.of(config.referenceName, new SourceInputFormatProvider(BigtableInputFormat.class, conf)));
   }
 
@@ -149,7 +156,9 @@ public void transform(KeyValue<ImmutableBytesWritable, Result> input, Emitter<St
           LOG.warn("Failed to process message, skipping it", e);
           break;
         case FAIL_PIPELINE:
-          throw new RuntimeException("Failed to process message", e);
+          String error = String.format("Failed to process message: %s", e.getMessage());
+          throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+            error, error, ErrorType.UNKNOWN, false, e);
         default:
           // this should never happen because it is validated at configure and prepare time
           throw new IllegalStateException(String.format("Unknown error handling strategy '%s'",

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

@@ -18,6 +18,7 @@
 
 import com.google.api.client.http.HttpResponseException;
 import com.google.common.base.Throwables;
+import io.cdap.cdap.api.data.format.UnexpectedFormatException;
 import io.cdap.cdap.api.exception.ErrorCategory;
 import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum;
 import io.cdap.cdap.api.exception.ErrorType;
@@ -59,18 +60,21 @@ public ProgramFailureException getExceptionDetails(Exception e, ErrorContext err
       if (t instanceof IllegalStateException) {
         return getProgramFailureException((IllegalStateException) t, errorContext);
       }
+      if (t instanceof UnexpectedFormatException) {
+        return getProgramFailureException((UnexpectedFormatException) t, errorContext);
+      }
     }
     return null;
   }
 
   /**
    * Get a ProgramFailureException with the given error
-   * information from {@link IllegalArgumentException}.
+   * information from {@link IllegalArgumentException}, {@link UnexpectedFormatException}.
    *
    * @param e The IllegalArgumentException to get the error information from.
    * @return A ProgramFailureException with the given error information.
    */
-  private ProgramFailureException getProgramFailureException(IllegalArgumentException e,
+  private ProgramFailureException getProgramFailureException(Exception e,
       ErrorContext errorContext) {
     String errorMessage = e.getMessage();
     return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.PLUGIN),
@@ -93,6 +97,8 @@ private ProgramFailureException getProgramFailureException(IllegalStateException
             e.getClass().getName(), errorMessage), ErrorType.SYSTEM, false, e);
   }
 
+
+
   /**
    * Get the external documentation link for the client errors if available.
    *

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 BIG_TABLE_SUPPORTED_DOC_URL = "https://cloud.google.com/bigtable/docs/status-codes";
 
   /**
    * Load a service account from the local file system.