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

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

@@ -28,6 +28,7 @@
 import io.cdap.plugin.http.common.http.OAuthUtil;
 
 import java.io.File;
+import java.util.Objects;
 import java.util.Optional;
 import javax.annotation.Nullable;
 
@@ -349,10 +350,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) && Objects.nonNull(waitTimeBetweenPages)
+                && 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 +368,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 +399,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 +416,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,120 @@
+/*
+ * Copyright © 2024 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 {
+    @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);
+            }
+            if (t instanceof IllegalStateException) {
+                return getProgramFailureException((IllegalStateException) t, errorContext);
+            }
+            if (t instanceof InvalidConfigPropertyException) {
+                return getProgramFailureException((InvalidConfigPropertyException) t, errorContext);
+            }
+            if (t instanceof NoSuchElementException) {
+                return getProgramFailureException((NoSuchElementException) t, errorContext);
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Get a ProgramFailureException with the given error
+     * information from {@link IllegalArgumentException}.
+     *
+     * @param e The IllegalArgumentException to get the error information from.
+     * @return A ProgramFailureException with the given error information.
+     */
+    private ProgramFailureException getProgramFailureException(IllegalArgumentException e, ErrorContext errorContext) {
+        String errorMessage = e.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.USER, false, e);
+    }
+
+    /**
+     * Get a ProgramFailureException with the given error
+     * information from {@link IllegalStateException}.
+     *
+     * @param e The IllegalStateException to get the error information from.
+     * @return A ProgramFailureException with the given error information.
+     */
+    private ProgramFailureException getProgramFailureException(IllegalStateException e, ErrorContext errorContext) {
+        String errorMessage = e.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.SYSTEM, false, e);
+    }
+
+    /**
+     * Get a ProgramFailureException with the given error
+     * information from {@link InvalidConfigPropertyException}.
+     *
+     * @param e The InvalidConfigPropertyException to get the error information from.
+     * @return A ProgramFailureException with the given error information.
+     */
+    private ProgramFailureException getProgramFailureException(InvalidConfigPropertyException e,
+                                                               ErrorContext errorContext) {
+        String errorMessage = e.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.USER, false, e);
+    }
+
+    /**
+     * Get a ProgramFailureException with the given error
+     * information from {@link NoSuchElementException}.
+     *
+     * @param e The NoSuchElementException to get the error information from.
+     * @return A ProgramFailureException with the given error information.
+     */
+    private ProgramFailureException getProgramFailureException(NoSuchElementException e, ErrorContext errorContext) {
+        String errorMessage = e.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.SYSTEM, false, e);
+    }
+}

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()) {

src/main/java/io/cdap/plugin/http/sink/batch/HTTPOutputFormat.java

@@ -37,12 +37,16 @@ public class HTTPOutputFormat extends OutputFormat<StructuredRecord, StructuredR
   static final String INPUT_SCHEMA_KEY = "http.sink.input.schema";
 
   @Override
-  public RecordWriter<StructuredRecord, StructuredRecord> getRecordWriter(TaskAttemptContext context)
-          throws IOException {
+  public RecordWriter<StructuredRecord, StructuredRecord> getRecordWriter(TaskAttemptContext context) {
     Configuration hConf = context.getConfiguration();
     HTTPSinkConfig config = GSON.fromJson(hConf.get(CONFIG_KEY), HTTPSinkConfig.class);
-    Schema inputSchema = Schema.parseJson(hConf.get(INPUT_SCHEMA_KEY));
-    return new HTTPRecordWriter(config, inputSchema);
+    Schema inputSchema;
+    try {
+      inputSchema = Schema.parseJson(hConf.get(INPUT_SCHEMA_KEY));
+      return new HTTPRecordWriter(config, inputSchema);
+    } catch (IOException e) {
+      throw new IllegalStateException("Unable to parse the input schema. Reason: " + e.getMessage(), e);
+    }
   }
 
   @Override