[PLUGIN-1826] Error Management HTTP Source and Sink and fix sonar issues

src/main/java/io/cdap/plugin/http/common/BaseHttpConfig.java

@@ -349,10 +349,16 @@ public void validate(FailureCollector failureCollector) {
         // Validate OAuth2 properties
         if (!containsMacro(PROPERTY_OAUTH2_ENABLED) && this.getOauth2Enabled()) {
             String reasonOauth2 = "OAuth2 is enabled";
-            assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2);
-            assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2);
-            assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2);
-            assertIsSet(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2);
+            assertIsSetWithFailureCollector(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2, failureCollector);
+            assertIsSetWithFailureCollector(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2, failureCollector);
+            assertIsSetWithFailureCollector(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2, failureCollector);
+            assertIsSetWithFailureCollector(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2, failureCollector);
+        }
+
+        if (!containsMacro(PROPERTY_WAIT_TIME_BETWEEN_PAGES) && waitTimeBetweenPages != null
+          && waitTimeBetweenPages < 0) {
+            failureCollector.addFailure("Wait Time Between Pages cannot be a negative number.",
+              null).withConfigProperty(PROPERTY_WAIT_TIME_BETWEEN_PAGES);
         }
 
         // Validate Authentication properties
@@ -361,16 +367,18 @@ public void validate(FailureCollector failureCollector) {
             case OAUTH2:
                 String reasonOauth2 = "OAuth2 is enabled";
                 if (!containsMacro(PROPERTY_TOKEN_URL)) {
-                    assertIsSet(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2);
+                    assertIsSetWithFailureCollector(getTokenUrl(), PROPERTY_TOKEN_URL, reasonOauth2, failureCollector);
                 }
                 if (!containsMacro(PROPERTY_CLIENT_ID)) {
-                    assertIsSet(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2);
+                    assertIsSetWithFailureCollector(getClientId(), PROPERTY_CLIENT_ID, reasonOauth2, failureCollector);
                 }
                 if (!containsMacro((PROPERTY_CLIENT_SECRET))) {
-                    assertIsSet(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2);
+                    assertIsSetWithFailureCollector(getClientSecret(), PROPERTY_CLIENT_SECRET, reasonOauth2,
+                      failureCollector);
                 }
                 if (!containsMacro(PROPERTY_REFRESH_TOKEN)) {
-                    assertIsSet(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2);
+                    assertIsSetWithFailureCollector(getRefreshToken(), PROPERTY_REFRESH_TOKEN, reasonOauth2,
+                      failureCollector);
                 }
                 break;
             case SERVICE_ACCOUNT:
@@ -390,10 +398,12 @@ public void validate(FailureCollector failureCollector) {
             case BASIC_AUTH:
                 String reasonBasicAuth = "Basic Authentication is enabled";
                 if (!containsMacro(PROPERTY_USERNAME)) {
-                    assertIsSet(getUsername(), PROPERTY_USERNAME, reasonBasicAuth);
+                    assertIsSetWithFailureCollector(getUsername(), PROPERTY_USERNAME, reasonBasicAuth,
+                            failureCollector);
                 }
                 if (!containsMacro(PROPERTY_PASSWORD)) {
-                    assertIsSet(getPassword(), PROPERTY_PASSWORD, reasonBasicAuth);
+                    assertIsSetWithFailureCollector(getPassword(), PROPERTY_PASSWORD, reasonBasicAuth,
+                            failureCollector);
                 }
                 break;
         }
@@ -405,4 +415,12 @@ public static void assertIsSet(Object propertyValue, String propertyName, String
                     String.format("Property '%s' must be set, since %s", propertyName, reason), propertyName);
         }
     }
+
+    public static void assertIsSetWithFailureCollector(Object propertyValue, String propertyName, String reason,
+                                                       FailureCollector failureCollector) {
+        if (propertyValue == null) {
+            failureCollector.addFailure(String.format("Property '%s' must be set, since %s", propertyName, reason),
+              null).withConfigProperty(propertyName);
+        }
+    }
 }

src/main/java/io/cdap/plugin/http/common/HttpErrorDetailsProvider.java

@@ -0,0 +1,93 @@
+/*
+ * 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.http.common;
+
+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.etl.api.exception.ErrorContext;
+import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
+import io.cdap.cdap.etl.api.validation.InvalidConfigPropertyException;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/**
+ * Error details provided for the HTTP
+ **/
+public class HttpErrorDetailsProvider implements ErrorDetailsProvider {
+
+  /**
+   * Supported document URL.
+   */
+  private static final String SUPPORTED_DOCUMENT_URL = "https://datatracker.ietf.org/doc/html/rfc7231#section-6";
+
+  /**
+   * Retrieves the supported document URL.
+   *
+   * @return the supported document URL
+   */
+  public static String getSupportedDocumentUrl() {
+    return SUPPORTED_DOCUMENT_URL;
+  }
+
+  @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 IllegalArgumentException) {
+        return getProgramFailureException((IllegalArgumentException) t, errorContext,
+            ErrorType.USER);
+      }
+      if (t instanceof IllegalStateException) {
+        return getProgramFailureException((IllegalStateException) t, errorContext,
+            ErrorType.SYSTEM);
+      }
+      if (t instanceof InvalidConfigPropertyException) {
+        return getProgramFailureException((InvalidConfigPropertyException) t, errorContext,
+            ErrorType.USER);
+      }
+      if (t instanceof NoSuchElementException) {
+        return getProgramFailureException((NoSuchElementException) t, errorContext,
+            ErrorType.SYSTEM);
+      }
+    }
+    return null;
+  }
+
+  /**
+   * Get a ProgramFailureException with the given error information from {@link Exception}.
+   *
+   * @param exception The IllegalArgumentException to get the error information from.
+   * @return A ProgramFailureException with the given error information.
+   */
+  private ProgramFailureException getProgramFailureException(Exception exception,
+      ErrorContext errorContext, ErrorType errorType) {
+    String errorMessage = exception.getMessage();
+    String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s";
+    return ErrorUtils.getProgramFailureException(
+        new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage,
+        String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), errorType, false,
+        exception);
+  }
+}

src/main/java/io/cdap/plugin/http/common/pagination/BaseHttpPaginationIterator.java

@@ -15,6 +15,11 @@
  */
 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.HttpErrorDetailsProvider;
 import io.cdap.plugin.http.common.RetryPolicy;
 import io.cdap.plugin.http.common.error.ErrorHandling;
 import io.cdap.plugin.http.common.error.HttpErrorHandler;
@@ -80,12 +85,30 @@ public BaseHttpPaginationIterator(BaseHttpSourceConfig config, PaginationIterato
   protected abstract String getNextPageUrl(HttpResponse httpResponse, BasePage page);
   public abstract boolean supportsSkippingPages();
 
-  protected boolean visitPageAndCheckStatusCode() throws IOException {
+  protected boolean visitPageAndCheckStatusCode() {
     if (response != null) { // close previous response
-      response.close();
+      try {
+        response.close();
+      } catch (IOException e) {
+        String errorReason = String.format("Failed to close response from '%s'", currentPageUrl);
+        String errorMessage = String.format("Failed to close response from '%s' with message: %s",
+            currentPageUrl, e.getMessage());
+        throw ErrorUtils.getProgramFailureException(
+            new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason, errorMessage,
+            ErrorType.SYSTEM, true, e);
+      }
     }
 
-    response = new HttpResponse(getHttpClient().executeHTTP(nextPageUrl));
+    try {
+      response = new HttpResponse(getHttpClient().executeHTTP(nextPageUrl));
+    } catch (IOException e) {
+      String errorMessage = String.format("Failed to execute request to '%s' with message: %s",
+          nextPageUrl, e.getMessage());
+      String errorReason = String.format("Failed to execute request to '%s'", nextPageUrl);
+      throw ErrorUtils.getProgramFailureException(
+          new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason, errorMessage,
+          ErrorType.SYSTEM, true, e);
+    }
     currentPageUrl = nextPageUrl;
     httpStatusCode = response.getStatusCode();
     RetryableErrorHandling errorHandlingStrategy = httpErrorHandler.getErrorHandlingStrategy(httpStatusCode);
@@ -94,7 +117,7 @@ protected boolean visitPageAndCheckStatusCode() throws IOException {
   }
 
   @Nullable
-  protected BasePage getNextPage() throws IOException {
+  protected BasePage getNextPage() {
     // no more pages
     if (nextPageUrl == null) {
       return null;
@@ -122,8 +145,18 @@ 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()));
+        ErrorUtils.ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(httpStatusCode);
+        String errorReason = String.format(
+            "Unable to read new page: %s. %s. For more details, see %s", httpStatusCode,
+            pair.getCorrectiveAction(), HttpErrorDetailsProvider.getSupportedDocumentUrl());
+        String errorMessage = String.format(
+            "Retry failed! Unable to read new page and execute request. "
+                + "Fetching from '%s' returned http error status code '%s'.", config.getUrl(),
+            httpStatusCode);
+        throw ErrorUtils.getProgramFailureException(
+            new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason, errorMessage,
+            pair.getErrorType(), true, ErrorCodeType.HTTP, String.valueOf(httpStatusCode),
+            HttpErrorDetailsProvider.getSupportedDocumentUrl(), null);
       case SKIP:
       case SEND:
         if (!this.supportsSkippingPages()) {
@@ -156,16 +189,11 @@ public String getCurrentPageUrl() {
    * False if no more pages to load or the page loaded has no elements.
    */
   protected boolean ensurePageIterable() {
-    try {
-      if (currentPageReturned) {
-        page = getNextPage();
-        currentPageReturned = false;
-      }
-
-      return page != null && page.hasNext(); // check hasNext() to stop on first empty page.
-    } catch (IOException e) {
-      throw new IllegalStateException("Failed to the load page", e);
+    if (currentPageReturned) {
+      page = getNextPage();
+      currentPageReturned = false;
     }
+    return page != null && page.hasNext(); // check hasNext() to stop on first empty page.
   }
 
   // for testing purposes
@@ -175,9 +203,9 @@ public HttpClient getHttpClient() {
 
   // for testing purposes
   BasePage createPageInstance(BaseHttpSourceConfig config, HttpResponse httpResponse,
-                              ErrorHandling postRetryStrategy) throws IOException {
+                              ErrorHandling postRetryStrategy)  {
     return PageFactory.createInstance(config, httpResponse, httpErrorHandler,
-                                      !postRetryStrategy.equals(ErrorHandling.SUCCESS));
+        !postRetryStrategy.equals(ErrorHandling.SUCCESS));
   }
 
   public PaginationIteratorState getCurrentState() {
@@ -204,14 +232,34 @@ public boolean hasNext() {
   }
 
   @Override
-  public void close() throws IOException {
+  public void close() {
     try {
       if (getHttpClient() != null) {
         getHttpClient().close();
       }
+    } catch (IOException e) {
+      ErrorUtils.ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(httpStatusCode);
+      String errorReason = String.format(
+          "Failed to close http client: %s. %s. For more details, see %s", httpStatusCode,
+          pair.getCorrectiveAction(), HttpErrorDetailsProvider.getSupportedDocumentUrl());
+      String errorMessage = String.format(
+          "Failed to close http client with status code %s with message %s.", httpStatusCode,
+          e.getMessage());
+      throw ErrorUtils.getProgramFailureException(
+          new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorReason, errorMessage,
+          pair.getErrorType(), true, ErrorCodeType.HTTP, String.valueOf(httpStatusCode),
+          HttpErrorDetailsProvider.getSupportedDocumentUrl(), e);
     } finally {
       if (response != null) {
-        response.close();
+        try {
+          response.close();
+        } catch (IOException e) {
+          String errorMessage = String.format("Failed to close http response with message: %s.",
+              e.getMessage());
+          throw ErrorUtils.getProgramFailureException(
+              new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
+              ErrorType.SYSTEM, true, e);
+        }
       }
     }
   }

src/main/java/io/cdap/plugin/http/common/pagination/CustomPaginationIterator.java

@@ -55,7 +55,7 @@ public boolean supportsSkippingPages() {
   }
 
   @Override
-  public void close() throws IOException {
+  public void close() {
     try {
       super.close();
     } finally {

src/main/java/io/cdap/plugin/http/common/pagination/page/JsonPage.java

@@ -59,16 +59,15 @@ class JsonPage extends BasePage {
     } else {
       this.insideElementJsonPathPart = config.getResultPath() == null ? "" : config.getResultPath();
     }
-
-    if (jsonElement.isJsonArray()) {
+    if (jsonElement != null && jsonElement.isJsonArray()) {
       this.iterator = jsonElement.getAsJsonArray().iterator();
-    } else if (jsonElement.isJsonObject()) {
+    } else if (jsonElement != null && jsonElement.isJsonObject()) {
       this.iterator = Collections.singleton(jsonElement).iterator();
     } else {
-      throw new IllegalArgumentException(String.format("Element found by '%s' json path is expected to be an object " +
-                                                         "or an array. Primitive found", config.getResultPath()));
+      throw new IllegalArgumentException(
+        String.format("Element found by '%s' json path is expected to be an object or an array. Primitive found",
+          config.getResultPath()));
     }
-
     this.fieldsMapping = config.getFullFieldsMapping();
   }