Support backward compatibility

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClient.java

@@ -61,9 +61,11 @@ public JavaNetSinkHttpClient(
         ComposeHttpStatusCodeCheckerConfig checkerConfig =
             ComposeHttpStatusCodeCheckerConfig.builder()
                 .properties(properties)
-                .errorWhiteListPrefix(HTTP_ERROR_SINK_CODE_WHITE_LIST)
-                .errorCodePrefix(HTTP_ERROR_SINK_CODES_LIST)
-                .retryableWhiteListPrefix("")   // TODO: sink retries not implemented yet
+                .deprecatedErrorWhiteListPrefix(HTTP_ERROR_SINK_CODE_WHITE_LIST)
+                .deprecatedCodePrefix(HTTP_ERROR_SINK_CODES_LIST)
+                .errorWhiteListPrefix("")      // TODO: sink not refactored yet
+                .errorCodePrefix("")
+                .retryableWhiteListPrefix("")
                 .retryableCodePrefix("")
                 .build();
 

src/main/java/com/getindata/connectors/http/internal/status/ComposeHttpStatusCodeChecker.java

@@ -10,7 +10,7 @@
 import lombok.Data;
 import lombok.RequiredArgsConstructor;
 import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.StringUtils;
+import static org.apache.flink.util.StringUtils.isNullOrWhitespaceOnly;
 
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 
@@ -35,8 +35,7 @@ public class ComposeHttpStatusCodeChecker implements HttpStatusCodeChecker {
 
     public ComposeHttpStatusCodeChecker(ComposeHttpStatusCodeCheckerConfig config) {
         // Handle deprecated configuration for backward compatibility.
-        if (!StringUtils.isNullOrWhitespaceOnly(config.getDeprecatedCodePrefix()) ||
-            !StringUtils.isNullOrWhitespaceOnly(config.getDeprecatedErrorWhiteListPrefix())) {
+        if (areDeprecatedPropertiesUsed(config)) {
             notRetryableErrorStatusCodes = buildPredicate(config, config.getDeprecatedCodePrefix(),
                 config.getDeprecatedErrorWhiteListPrefix(), DEFAULT_DEPRECATED_ERROR_CODES);
             retryableErrorStatusCodes = integer -> false;
@@ -48,15 +47,28 @@ public ComposeHttpStatusCodeChecker(ComposeHttpStatusCodeCheckerConfig config) {
         }
     }
 
+    private boolean areDeprecatedPropertiesUsed(ComposeHttpStatusCodeCheckerConfig config) {
+        boolean whiteListDefined =
+            !isNullOrWhitespaceOnly(config.getDeprecatedErrorWhiteListPrefix());
+        boolean codeListDefined = !isNullOrWhitespaceOnly(config.getDeprecatedCodePrefix());
+
+        return (whiteListDefined && !isNullOrWhitespaceOnly(
+            config.getProperties().getProperty(config.getDeprecatedErrorWhiteListPrefix())))
+            || (codeListDefined && !isNullOrWhitespaceOnly(
+            config.getProperties().getProperty(config.getDeprecatedCodePrefix())));
+    }
+
     private Predicate<Integer> buildPredicate(
         ComposeHttpStatusCodeCheckerConfig config,
         String errorCodePrefix,
         String whiteListPrefix,
         Predicate<Integer> defaultErrorCodes) {
         Properties properties = config.getProperties();
 
-        String errorCodes = properties.getProperty(errorCodePrefix, "");
-        String whitelistCodes = properties.getProperty(whiteListPrefix, "");
+        String errorCodes =
+            errorCodePrefix == null ? "" : properties.getProperty(errorCodePrefix, "");
+        String whitelistCodes =
+            whiteListPrefix == null ? "" : properties.getProperty(whiteListPrefix, "");
 
         Predicate<Integer> errorPredicate =
             prepareErrorCodes(errorCodes).orElse(defaultErrorCodes);
@@ -74,7 +86,7 @@ private Predicate<Integer> buildPredicate(
      */
     private Optional<Predicate<Integer>> prepareErrorCodes(String statusCodesStr) {
         return Arrays.stream(statusCodesStr.split(HttpConnectorConfigConstants.PROP_DELIM))
-            .filter(code -> !StringUtils.isNullOrWhitespaceOnly(code))
+            .filter(code -> !isNullOrWhitespaceOnly(code))
             .map(code -> code.toUpperCase().trim())
             .map(codeStr -> {
                 Preconditions.checkArgument(

src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java

@@ -7,10 +7,6 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Optional;
-import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_NON_RETRYABLE_SOURCE_LOOKUP_CODES_LIST;
-import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_NON_RETRYABLE_SOURCE_LOOKUP_CODE_WHITE_LIST;
-import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_SOURCE_LOOKUP_CODES_LIST;
-import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_SOURCE_LOOKUP_CODE_WHITE_LIST;
 import static java.lang.String.format;
 
 import lombok.AllArgsConstructor;
@@ -32,8 +28,12 @@
 import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker.ComposeHttpStatusCodeCheckerConfig;
 import com.getindata.connectors.http.internal.status.HttpResponseStatus;
 import com.getindata.connectors.http.internal.status.HttpStatusCodeChecker;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_NON_RETRYABLE_SOURCE_LOOKUP_CODES_LIST;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_NON_RETRYABLE_SOURCE_LOOKUP_CODE_WHITE_LIST;
 import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_RETRYABLE_SOURCE_LOOKUP_CODES_LIST;
 import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_RETRYABLE_SOURCE_LOOKUP_CODE_WHITE_LIST;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_SOURCE_LOOKUP_CODES_LIST;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.HTTP_ERROR_SOURCE_LOOKUP_CODE_WHITE_LIST;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.LOOKUP_RESTART_STRATEGY_EXPONENTIAL_DELAY_ATTEMPTS;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.LOOKUP_RESTART_STRATEGY_EXPONENTIAL_DELAY_INITIAL_DELAY;
 import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.LOOKUP_RESTART_STRATEGY_EXPONENTIAL_DELAY_MAX_DELAY;

src/test/java/com/getindata/connectors/http/internal/sink/HttpSinkConnectionTest.java

@@ -274,7 +274,6 @@ public void testFailedConnection404OnWhiteList() throws Exception {
                 .setSinkHttpClientBuilder(JavaNetSinkHttpClient::new)
                 .setProperty("gid.connector.http.sink.error.code.exclude", "404, 405")
                 .setProperty("gid.connector.http.sink.error.code", "4XX")
-                .setProperty("gid.connector.http.sink.error-retryable.code.exclude", "404, 405")
                 .build();
         source.sinkTo(httpSink);
         env.execute("Http Sink test failed connection");

src/test/java/com/getindata/connectors/http/internal/status/ComposeHttpStatusCodeCheckerTest.java

@@ -17,10 +17,11 @@
 
 class ComposeHttpStatusCodeCheckerTest {
 
-    private static final String NOT_RETRYABLE_CODE_PROPERTY = "error.code";
-    private static final String NOT_RETRYABLE_WHITELIST_PROPERTY = "error.code.exclude";
-    private static final String RETRYABLE_CODE_PROPERTY = "retryable.code";
-    private static final String RETRYABLE_WHITELIST_PROPERTY = "retryable.code.exclude";
+    private static final String NOT_RETRYABLE_CODE_PROPERTY = "error.non-retryable.code";
+    private static final String NOT_RETRYABLE_WHITELIST_PROPERTY =
+        "error.non-retryable.code.exclude";
+    private static final String RETRYABLE_CODE_PROPERTY = "error.retryable.code";
+    private static final String RETRYABLE_WHITELIST_PROPERTY = "error.retryable.code.exclude";
 
     @Test
     void shouldReturnAppropriateStatusByDefault() {