Raise exception when error processing the input stream in JSON extractor

linkedin · Dec 2, 2021 · df2ef79 · df2ef79
2 parents 028c471 + 35ed330
commit df2ef79
Show file tree

Hide file tree

Showing 5 changed files with 1 addition and 5 deletions.
diff --git a/cdi-core/src/main/java/com/linkedin/cdi/extractor/JsonExtractor.java b/cdi-core/src/main/java/com/linkedin/cdi/extractor/JsonExtractor.java
@@ -225,7 +225,7 @@ protected boolean processInputStream(long starting) {
     } catch (Exception e) {
       LOG.error("Source Error: {}", e.getMessage());
       state.setWorkingState(WorkUnitState.WorkingState.FAILED);
-      return false;
+      throw new RuntimeException(e);
     }
 
     LOG.debug("Checking parsed Json object");

diff --git a/cdi-core/src/main/java/com/linkedin/cdi/keys/HttpKeys.java b/cdi-core/src/main/java/com/linkedin/cdi/keys/HttpKeys.java
@@ -29,7 +29,6 @@ public class HttpKeys extends JobKeys {
 
   @Override
   public void logDebugAll() {
-    super.logDebugAll();
     LOG.debug("These are values in HttpSource");
     LOG.debug("Http Request Headers: {}", httpRequestHeaders);
     //LOG.debug("Http Request Headers with Authentication: {}", httpRequestHeadersWithAuthentication.toString());

diff --git a/cdi-core/src/main/java/com/linkedin/cdi/keys/JdbcKeys.java b/cdi-core/src/main/java/com/linkedin/cdi/keys/JdbcKeys.java
@@ -24,7 +24,6 @@ public class JdbcKeys extends JobKeys {
 
   @Override
   public void logDebugAll() {
-    super.logDebugAll();
     LOG.debug("These are values in JdbcSource");
     LOG.debug("JDBC statement: {}", jdbcStatement);
     LOG.debug("Initial values of dynamic parameters: {}", initialParameterValues);

diff --git a/cdi-core/src/main/java/com/linkedin/cdi/keys/S3Keys.java b/cdi-core/src/main/java/com/linkedin/cdi/keys/S3Keys.java
@@ -23,7 +23,6 @@ public class S3Keys extends JobKeys {
 
   @Override
   public void logDebugAll() {
-    super.logDebugAll();
     LOG.debug("These are values in S3SourceV2:");
     LOG.debug("S3 Bucket: {}", bucket);
     LOG.debug("S3 endpoint: {}", endpoint);

diff --git a/cdi-core/src/main/java/com/linkedin/cdi/keys/SftpKeys.java b/cdi-core/src/main/java/com/linkedin/cdi/keys/SftpKeys.java
@@ -18,7 +18,6 @@ public class SftpKeys extends JobKeys {
 
   @Override
   public void logDebugAll() {
-    super.logDebugAll();
     LOG.debug("These are values in SftpSource:");
     LOG.debug("sftp source path: {}", filesPath);
     LOG.debug("path separator: {}", pathSeparator);