Add error details provider for Amazon S3 plugin

Author: itsankit-google

pom.xml

@@ -142,6 +142,11 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-aws</artifactId>
+      <version>${hadoop.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-data-pipeline3_2.12</artifactId>
@@ -283,6 +288,11 @@
       <artifactId>aws-java-sdk-s3</artifactId>
       <version>${aws.sdk.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>3.12.0</version>
+    </dependency>
     <dependency>
       <groupId>com.amazonaws</groupId>
       <artifactId>aws-java-sdk-sts</artifactId>
@@ -310,7 +320,11 @@
           <extensions>true</extensions>
           <configuration>
             <instructions>
-              <_exportcontents>io.cdap.plugin.aws.s3.*</_exportcontents>
+              <_exportcontents>
+                io.cdap.plugin.aws.s3.*;
+                org.apache.hadoop.fs.s3a.*;
+                org.apache.hadoop.fs.s3native.*;
+              </_exportcontents>
               <Embed-Dependency>*;inline=false;scope=compile</Embed-Dependency>
               <Embed-Transitive>true</Embed-Transitive>
               <Embed-Directory>lib</Embed-Directory>
@@ -492,7 +506,11 @@
             <extensions>true</extensions>
             <configuration>
               <instructions>
-                <_exportcontents>io.cdap.plugin.aws.s3.*</_exportcontents>
+                <_exportcontents>
+                  io.cdap.plugin.aws.s3.*;
+                  org.apache.hadoop.fs.s3a.*;
+                  org.apache.hadoop.fs.s3native.*;
+                </_exportcontents>
                 <Embed-Dependency>*;inline=false;scope=compile</Embed-Dependency>
                 <Embed-Transitive>true</Embed-Transitive>
                 <Embed-Directory>lib</Embed-Directory>

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

@@ -0,0 +1,58 @@
+/*
+ * 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.AmazonServiceException;
+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 final 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 AmazonServiceException) {
+        AmazonServiceException s3Exception = (AmazonServiceException) t;
+        String errorMessage = s3Exception.getErrorMessage();
+        String errorMessageWithDetails = s3Exception.getMessage();
+        return ErrorUtils.getProgramFailureException(
+            new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN, s3Exception.getErrorCode()),
+            errorMessage, errorMessageWithDetails, ErrorType.USER, true, ErrorCodeType.HTTP,
+            String.valueOf(s3Exception.getStatusCode()), S3_EXTERNAL_DOC, t);
+      }
+    }
+    return null;
+  }
+}

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

@@ -31,6 +31,7 @@
 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.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;
@@ -87,6 +88,11 @@ protected LineageRecorder getLineageRecorder(BatchSinkContext context) {
     return new LineageRecorder(context, asset);
   }
 
+  @Override
+  protected String getErrorDetailsProviderClassName() {
+    return AmazonErrorDetailsProvider.class.getName();
+  }
+
   @Override
   protected Map<String, String> getFileSystemProperties(BatchSinkContext context) {
     // when context is null, it is configure time, by that time, it will always use s3n

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

@@ -30,6 +30,7 @@
 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.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;
@@ -89,6 +90,11 @@ public void prepareRun(BatchSourceContext context) throws Exception {
     super.prepareRun(context);
   }
 
+  @Override
+  protected String getErrorDetailsProviderClassName() {
+    return AmazonErrorDetailsProvider.class.getName();
+  }
+
   @Override
   protected LineageRecorder getLineageRecorder(BatchSourceContext context) {
     return new LineageRecorder(context, asset);