Adding User and System Errors in FileErrorDetailsProvider

Author: AnkitCLI

core-plugins/src/main/java/io/cdap/plugin/batch/source/FileErrorDetailsProvider.java

@@ -16,11 +16,131 @@
 
 package io.cdap.plugin.batch.source;
 
+import com.google.common.base.Throwables;
+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.api.exception.ProgramFailureException;
+import io.cdap.cdap.api.metadata.MetadataException;
+import io.cdap.cdap.etl.api.exception.ErrorContext;
 import io.cdap.plugin.common.HydratorErrorDetailsProvider;
+import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.apache.hadoop.fs.InvalidPathException;
+import org.apache.hadoop.fs.InvalidRequestException;
+import org.apache.hadoop.fs.ParentNotDirectoryException;
+import org.apache.hadoop.fs.PathIsNotDirectoryException;
+import org.apache.hadoop.hdfs.BlockMissingException;
+import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
+import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
+import org.apache.hadoop.hdfs.server.datanode.ReplicaNotFoundException;
+import org.apache.hadoop.hdfs.server.namenode.SafeModeException;
+import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.ipc.StandbyException;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.util.DiskChecker;
+
+import java.io.FileNotFoundException;
+import java.net.NoRouteToHostException;
+import java.net.SocketTimeoutException;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
+import javax.security.auth.login.FailedLoginException;
 
 /**
  * FileErrorDetails provider
  */
 public class FileErrorDetailsProvider extends HydratorErrorDetailsProvider {
-  
+  static final String ERROR_MESSAGE_FORMAT = "Error occurred in the phase: '%s'. %s: %s";
+
+  @Override
+  public ProgramFailureException getExceptionDetails(Exception e, ErrorContext errorContext) {
+    // Call super method to get base exception details
+    ProgramFailureException ex = super.getExceptionDetails(e, errorContext);
+    if (ex != null) {
+      return ex;
+    }
+
+    // Use the new function to handle file-based program failure exceptions
+    return getFileBasedExceptionDetails(e, errorContext);
+  }
+
+  public ProgramFailureException getFileBasedExceptionDetails(Exception e, ErrorContext errorContext) {
+    List<Throwable> causalChain = Throwables.getCausalChain(e);
+    for (Throwable t : causalChain) {
+      if (t instanceof FileNotFoundException) {
+        return getFileBasedProgramFailureException((FileNotFoundException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof AccessControlException) {
+        return getFileBasedProgramFailureException((AccessControlException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof ParentNotDirectoryException) {
+        return getFileBasedProgramFailureException((ParentNotDirectoryException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof InvalidPathException) {
+        return getFileBasedProgramFailureException((InvalidPathException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof FileAlreadyExistsException) {
+        return getFileBasedProgramFailureException((FileAlreadyExistsException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof QuotaExceededException) {
+        return getFileBasedProgramFailureException((QuotaExceededException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof PathIsNotDirectoryException) {
+        return getFileBasedProgramFailureException((PathIsNotDirectoryException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof InvalidRequestException) {
+        return getFileBasedProgramFailureException((InvalidRequestException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof ChecksumException) {
+        return getFileBasedProgramFailureException((ChecksumException) t, errorContext, ErrorType.USER);
+      }
+      if (t instanceof RemoteException) {
+        return getFileBasedProgramFailureException((RemoteException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof SocketTimeoutException) {
+        return getFileBasedProgramFailureException((SocketTimeoutException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof DiskChecker.DiskOutOfSpaceException) {
+        return getFileBasedProgramFailureException((DiskChecker.DiskOutOfSpaceException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof StandbyException) {
+        return getFileBasedProgramFailureException((StandbyException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof NoRouteToHostException) {
+        return getFileBasedProgramFailureException((NoRouteToHostException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof BlockMissingException) {
+        return getFileBasedProgramFailureException((BlockMissingException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof ReplicaNotFoundException) {
+        return getFileBasedProgramFailureException((ReplicaNotFoundException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof InvalidBlockTokenException) {
+        return getFileBasedProgramFailureException((InvalidBlockTokenException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof SafeModeException) {
+        return getFileBasedProgramFailureException((SafeModeException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof TimeoutException) {
+        return getFileBasedProgramFailureException((TimeoutException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof FailedLoginException) {
+        return getFileBasedProgramFailureException((FailedLoginException) t, errorContext, ErrorType.SYSTEM);
+      }
+      if (t instanceof MetadataException) {
+        return getFileBasedProgramFailureException((MetadataException) t, errorContext, ErrorType.SYSTEM);
+      }
+    }
+    return null;
+  }
+
+  public ProgramFailureException getFileBasedProgramFailureException(Exception e, @Nullable ErrorContext errorContext,
+                                                                     ErrorType errorType) {
+    String errorMessage = e.getMessage();
+    return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+    errorMessage, errorContext != null ? String.format(ERROR_MESSAGE_FORMAT, errorContext.getPhase(), e.getClass()
+    .getName(), errorMessage) : String.format("%s: %s", e.getClass().getName(), errorMessage), errorType, false, e);
+  }
 }