reformat

Author: maciejmaciejko-gid

README.md

@@ -675,6 +675,7 @@ The mapping from Http Json Response to SQL table schema is done via Flink's Json
     instead `HttpSinkRequestEntry`.
 - Version 0.20
   - Http source table parameters: `gid.connector.http.source.lookup.error.code` and `gid.connector.http.source.lookup.error.code.exclude` were removed. These parameters described http status codes which was silently ignored by source lookup table (logged only). it's not recommended to ignore all error response but it's still possible. To do this set all codes as success: `'gid.connector.http.source.lookup.success-codes' = '1XX,2XX,3XX,4XX,5XX'` and ignore body from the others responses than 200s: `'gid.connector.http.source.lookup.ignored-response-codes' = '1XX,3XX,4XX,5XX'`. You can still exclude some error codes marking it as transition errors - `gid.connector.http.source.lookup.retry-codes`. Retry-codes have to be excluded from both `success-codes` and `ignored-response-codes`.
+  - Added dependency io.github.resilience4j:resilience4j-retry
 
 ## TODO
 

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

@@ -36,11 +36,11 @@ public class HttpLookupTableSourceFactory implements DynamicTableSourceFactory {
 
     private static DataTypes.Field columnToField(Column column) {
         return FIELD(
-                column.getName(),
-                // only a column in a schema should have a time attribute,
-                // a field should not propagate the attribute because it might be used in a
-                // completely different context
-                removeTimeAttribute(column.getDataType()));
+            column.getName(),
+            // only a column in a schema should have a time attribute,
+            // a field should not propagate the attribute because it might be used in a
+            // completely different context
+            removeTimeAttribute(column.getDataType()));
     }
 
     public static DataType row(List<Field> fields) {
@@ -50,36 +50,36 @@ public static DataType row(List<Field> fields) {
     @Override
     public DynamicTableSource createDynamicTableSource(Context dynamicTableContext) {
         FactoryUtil.TableFactoryHelper helper =
-                FactoryUtil.createTableFactoryHelper(this, dynamicTableContext);
+            FactoryUtil.createTableFactoryHelper(this, dynamicTableContext);
 
         ReadableConfig readable = helper.getOptions();
         helper.validateExcept(
-                // properties coming from org.apache.flink.table.api.config.ExecutionConfigOptions
-                "table.",
-                HttpConnectorConfigConstants.GID_CONNECTOR_HTTP,
-                LOOKUP_REQUEST_FORMAT.key()
+            // properties coming from org.apache.flink.table.api.config.ExecutionConfigOptions
+            "table.",
+            HttpConnectorConfigConstants.GID_CONNECTOR_HTTP,
+            LOOKUP_REQUEST_FORMAT.key()
         );
         validateHttpLookupSourceOptions(readable);
 
         DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
-                helper.discoverDecodingFormat(
-                        DeserializationFormatFactory.class,
-                        FactoryUtil.FORMAT
-                );
+            helper.discoverDecodingFormat(
+                    DeserializationFormatFactory.class,
+                    FactoryUtil.FORMAT
+            );
 
         HttpLookupConfig lookupConfig = getHttpLookupOptions(dynamicTableContext, readable);
 
         ResolvedSchema resolvedSchema = dynamicTableContext.getCatalogTable().getResolvedSchema();
 
         DataType physicalRowDataType =
-                toRowDataType(resolvedSchema.getColumns(), Column::isPhysical);
+            toRowDataType(resolvedSchema.getColumns(), Column::isPhysical);
 
         return new HttpLookupTableSource(
-                physicalRowDataType,
-                lookupConfig,
-                decodingFormat,
-                dynamicTableContext,
-                getLookupCache(readable)
+            physicalRowDataType,
+            lookupConfig,
+            decodingFormat,
+            dynamicTableContext,
+            getLookupCache(readable)
         );
     }
 
@@ -89,7 +89,7 @@ protected void validateHttpLookupSourceOptions(ReadableConfig tableOptions)
         tableOptions.getOptional(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL).ifPresent(url -> {
             if (tableOptions.getOptional(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST).isEmpty()) {
                 throw new IllegalArgumentException("Config option " +
-                        SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST.key() + " is required, if " +
+                    SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST.key() + " is required, if " +
                         SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL.key() + " is configured.");
             }
         });
@@ -108,57 +108,57 @@ public Set<ConfigOption<?>> requiredOptions() {
     @Override
     public Set<ConfigOption<?>> optionalOptions() {
         return Set.of(
-                URL_ARGS,
-                ASYNC_POLLING,
-                LOOKUP_METHOD,
-                REQUEST_CALLBACK_IDENTIFIER,
-
-                LookupOptions.CACHE_TYPE,
-                LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_ACCESS,
-                LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_WRITE,
-                LookupOptions.PARTIAL_CACHE_MAX_ROWS,
-                LookupOptions.PARTIAL_CACHE_CACHE_MISSING_KEY,
-
-                SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION,
-                SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
-                SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL,
-
-                LookupOptions.MAX_RETRIES,
-                SOURCE_LOOKUP_RETRY_STRATEGY,
-                SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY,
-                SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF,
-                SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER,
-                SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF,
-
-                SOURCE_LOOKUP_HTTP_SUCCESS_CODES,
-                SOURCE_LOOKUP_HTTP_RETRY_CODES,
-                SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
-
-                SOURCE_LOOKUP_CONNECTION_TIMEOUT        // TODO: add request timeout from properties
+            URL_ARGS,
+            ASYNC_POLLING,
+            LOOKUP_METHOD,
+            REQUEST_CALLBACK_IDENTIFIER,
+
+            LookupOptions.CACHE_TYPE,
+            LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_ACCESS,
+            LookupOptions.PARTIAL_CACHE_EXPIRE_AFTER_WRITE,
+            LookupOptions.PARTIAL_CACHE_MAX_ROWS,
+            LookupOptions.PARTIAL_CACHE_CACHE_MISSING_KEY,
+
+            SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION,
+            SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
+            SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL,
+
+            LookupOptions.MAX_RETRIES,
+            SOURCE_LOOKUP_RETRY_STRATEGY,
+            SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY,
+            SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF,
+            SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER,
+            SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF,
+
+            SOURCE_LOOKUP_HTTP_SUCCESS_CODES,
+            SOURCE_LOOKUP_HTTP_RETRY_CODES,
+            SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
+
+            SOURCE_LOOKUP_CONNECTION_TIMEOUT        // TODO: add request timeout from properties
         );
     }
 
     private HttpLookupConfig getHttpLookupOptions(Context context, ReadableConfig readableConfig) {
 
         Properties httpConnectorProperties =
-                ConfigUtils.getHttpConnectorProperties(context.getCatalogTable().getOptions());
+            ConfigUtils.getHttpConnectorProperties(context.getCatalogTable().getOptions());
 
         final HttpPostRequestCallbackFactory<HttpLookupSourceRequestEntry>
                 postRequestCallbackFactory =
-                FactoryUtil.discoverFactory(
+                    FactoryUtil.discoverFactory(
                         context.getClassLoader(),
                         HttpPostRequestCallbackFactory.class,
                         readableConfig.get(REQUEST_CALLBACK_IDENTIFIER)
-                );
+        );
 
         return HttpLookupConfig.builder()
-                .lookupMethod(readableConfig.get(LOOKUP_METHOD))
-                .url(readableConfig.get(URL))
-                .useAsync(readableConfig.get(ASYNC_POLLING))
-                .properties(httpConnectorProperties)
-                .readableConfig(readableConfig)
-                .httpPostRequestCallback(postRequestCallbackFactory.createHttpPostRequestCallback())
-                .build();
+            .lookupMethod(readableConfig.get(LOOKUP_METHOD))
+            .url(readableConfig.get(URL))
+            .useAsync(readableConfig.get(ASYNC_POLLING))
+            .properties(httpConnectorProperties)
+            .readableConfig(readableConfig)
+            .httpPostRequestCallback(postRequestCallbackFactory.createHttpPostRequestCallback())
+            .build();
     }
 
     @Nullable
@@ -177,13 +177,13 @@ private LookupCache getLookupCache(ReadableConfig tableOptions) {
     // Backport from Flink 1.15-Master
     private DataType toRowDataType(List<Column> columns, Predicate<Column> columnPredicate) {
         return columns.stream()
-                .filter(columnPredicate)
-                .map(HttpLookupTableSourceFactory::columnToField)
-                .collect(
-                        Collectors.collectingAndThen(Collectors.toList(),
-                                HttpLookupTableSourceFactory::row))
-                // the row should never be null
-                .notNull();
+            .filter(columnPredicate)
+            .map(HttpLookupTableSourceFactory::columnToField)
+            .collect(
+                    Collectors.collectingAndThen(Collectors.toList(),
+                            HttpLookupTableSourceFactory::row))
+            // the row should never be null
+            .notNull();
     }
 
     // Backport End