update runtime error

Author: psainics

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

@@ -25,6 +25,9 @@
 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;
@@ -153,8 +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);
-          // #TODO
+          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

@@ -61,20 +61,20 @@ public ProgramFailureException getExceptionDetails(Exception e, ErrorContext err
         return getProgramFailureException((IllegalStateException) t, errorContext);
       }
       if (t instanceof UnexpectedFormatException) {
-         // TODO
+        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),
@@ -97,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.
    *