Sink update

data-integrations · Dec 18, 2024 · 0af6ccc · 0af6ccc
1 parent 43ca97b
commit 0af6ccc
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 5 deletions.
diff --git a/pom.xml b/pom.xml
@@ -21,7 +21,7 @@
   <name>HTTP Plugins</name>
   <groupId>io.cdap</groupId>
   <artifactId>http-plugins</artifactId>
-  <version>1.5.4-SNAPSHOT</version>
+  <version>1.5.0-SNAPSHOT</version>
 
   <licenses>
     <license>

diff --git a/src/main/java/io/cdap/plugin/http/common/pagination/BaseHttpPaginationIterator.java b/src/main/java/io/cdap/plugin/http/common/pagination/BaseHttpPaginationIterator.java
@@ -15,6 +15,10 @@
  */
 package io.cdap.plugin.http.common.pagination;
 
+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.plugin.http.common.RetryPolicy;
 import io.cdap.plugin.http.common.error.ErrorHandling;
 import io.cdap.plugin.http.common.error.HttpErrorHandler;
@@ -26,6 +30,7 @@
 import io.cdap.plugin.http.common.pagination.state.PaginationIteratorState;
 import io.cdap.plugin.http.common.pagination.state.UrlPaginationIteratorState;
 import io.cdap.plugin.http.source.common.BaseHttpSourceConfig;
+import okhttp3.internal.http2.ErrorCode;
 import org.awaitility.Awaitility;
 import org.awaitility.core.ConditionTimeoutException;
 import org.awaitility.pollinterval.FixedPollInterval;
@@ -36,9 +41,12 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
@@ -122,8 +130,13 @@ protected BasePage getNextPage() throws IOException {
       case SUCCESS:
         break;
       case STOP:
-        throw new IllegalStateException(String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
-                                                      nextPageUrl, httpStatusCode, response.getBody()));
+        String errorMessage = String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
+                nextPageUrl, httpStatusCode, response.getBody());
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+                errorMessage, String.format("Error message: %s", errorMessage),
+                ErrorUtils.getActionErrorByStatusCode(httpStatusCode).getErrorType(),
+                true, ErrorCodeType.HTTP, String.valueOf(httpStatusCode),
+                "https://developer.mozilla.org/en-US/docs/Web/HTTP/Status", new IllegalStateException(errorMessage));
       case SKIP:
       case SEND:
         if (!this.supportsSkippingPages()) {

diff --git a/src/main/java/io/cdap/plugin/http/sink/batch/HTTPRecordWriter.java b/src/main/java/io/cdap/plugin/http/sink/batch/HTTPRecordWriter.java
@@ -21,6 +21,7 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 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.plugin.http.common.RetryPolicy;
@@ -383,8 +384,13 @@ private void flushMessageBuffer() {
       case SUCCESS:
         break;
       case STOP:
-        throw new IllegalStateException(String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
-                config.getUrl(), httpStatusCode, httpResponseBody));
+        String errorMessage = String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
+                config.getUrl(), httpStatusCode, httpResponseBody);
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
+                errorMessage, String.format("Error message: %s", errorMessage),
+                ErrorUtils.getActionErrorByStatusCode(httpStatusCode).getErrorType(),
+                true, ErrorCodeType.HTTP, String.valueOf(httpStatusCode),
+                "https://developer.mozilla.org/en-US/docs/Web/HTTP/Status", new IllegalStateException(errorMessage));
       case SKIP:
       case SEND:
         LOG.warn(String.format("Fetching from url '%s' returned status code '%d' and body '%s'",