Error management for Amazon S3 Source and Sink plugin

Author: psainics

src/main/java/io/cdap/plugin/aws/s3/common/AmazonErrorDetailsProvider.java

@@ -0,0 +1,56 @@
+/*
+ * 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.aws.s3.common;
+
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+import com.google.common.base.Throwables;
+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.api.exception.ProgramFailureException;
+import io.cdap.cdap.etl.api.exception.ErrorContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
+import java.util.List;
+
+/**
+ * Error details provided for the Amazon S3
+ **/
+public class AmazonErrorDetailsProvider implements ErrorDetailsProvider {
+
+  static final String S3_EXTERNAL_DOC = "https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html";
+
+  @Override
+  public ProgramFailureException getExceptionDetails(Exception e, ErrorContext errorContext) {
+    List<Throwable> causalChain = Throwables.getCausalChain(e);
+    for (Throwable t : causalChain) {
+      if (t instanceof ProgramFailureException) {
+        // if causal chain already has program failure exception, return null to avoid double wrap.
+        return null;
+      }
+      if (t instanceof AmazonS3Exception) {
+        AmazonS3Exception s3Exception = (AmazonS3Exception) t;
+        String errorMessage = s3Exception.getErrorMessage();
+        String errorMessageWithDetails = s3Exception.getMessage();
+        return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+          errorMessage, errorMessageWithDetails, ErrorType.USER, true, ErrorCodeType.HTTP, s3Exception.getErrorCode(),
+          S3_EXTERNAL_DOC, t);
+      }
+    }
+    return null;
+  }
+}

src/main/java/io/cdap/plugin/aws/s3/sink/S3BatchSink.java

@@ -31,6 +31,8 @@
 import io.cdap.cdap.etl.api.batch.BatchSink;
 import io.cdap.cdap.etl.api.batch.BatchSinkContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
+import io.cdap.plugin.aws.s3.common.AmazonErrorDetailsProvider;
 import io.cdap.plugin.aws.s3.common.S3ConnectorConfig;
 import io.cdap.plugin.aws.s3.common.S3Constants;
 import io.cdap.plugin.aws.s3.common.S3Path;
@@ -78,6 +80,10 @@ public void prepareRun(BatchSinkContext context) throws Exception {
     asset = Asset.builder(referenceName)
       .setFqn(config.getPath().replace(S3Path.SCHEME, "s3://")).build();
 
+    // set error details provider
+    context.setErrorDetailsProvider(
+        new ErrorDetailsProviderSpec(AmazonErrorDetailsProvider.class.getName()));
+
     // super is called down here to avoid instantiating the lineage recorder with a null asset
     super.prepareRun(context);
   }
@@ -207,8 +213,9 @@ public void validate(FailureCollector collector) {
         try {
           getFilesystemProperties();
         } catch (Exception e) {
-          collector.addFailure("File system properties must be a valid json.", null)
-            .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
+          collector.addFailure(String.format("File system properties must be a valid json, %s, %s",
+                  e.getClass().getName(), e.getMessage()), null)
+              .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
         }
       }
     }

src/main/java/io/cdap/plugin/aws/s3/source/S3BatchSource.java

@@ -30,6 +30,8 @@
 import io.cdap.cdap.etl.api.batch.BatchSource;
 import io.cdap.cdap.etl.api.batch.BatchSourceContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
+import io.cdap.plugin.aws.s3.common.AmazonErrorDetailsProvider;
 import io.cdap.plugin.aws.s3.common.S3ConnectorConfig;
 import io.cdap.plugin.aws.s3.common.S3Constants;
 import io.cdap.plugin.aws.s3.common.S3EmptyInputFormat;
@@ -85,6 +87,10 @@ public void prepareRun(BatchSourceContext context) throws Exception {
     asset = Asset.builder(referenceName)
       .setFqn(config.getPath().replace(S3Path.SCHEME, "s3://")).build();
 
+    // set error details provider
+    context.setErrorDetailsProvider(
+        new ErrorDetailsProviderSpec(AmazonErrorDetailsProvider.class.getName()));
+
     // super is called down here to avoid instantiating the lineage recorder with a null asset
     super.prepareRun(context);
   }
@@ -220,8 +226,9 @@ public void validate(FailureCollector collector) {
         try {
           getFilesystemProperties();
         } catch (Exception e) {
-          collector.addFailure("File system properties must be a valid json.", null)
-            .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
+          collector.addFailure(String.format("File system properties must be a valid json, %s: %s",
+                  e.getClass().getName(), e.getMessage()), null)
+              .withConfigProperty(NAME_FILE_SYSTEM_PROPERTIES).withStacktrace(e.getStackTrace());
         }
       }
     }