HTTP99 address review feedback

Signed-off-by: davidradl <[email protected]>

Author: davidradl

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

@@ -10,6 +10,8 @@
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 
+import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
@@ -23,8 +25,6 @@
 import org.apache.flink.types.Row;
 import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 
 import com.getindata.connectors.http.LookupArg;
 import com.getindata.connectors.http.LookupQueryCreator;
@@ -53,9 +53,10 @@
  * http/base/aaaa will be issued.
  *
  */
+@Slf4j
+@Builder
 public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
     private static final long serialVersionUID = 1L;
-    private static final Logger log = LogManager.getLogger(GenericJsonAndUrlQueryCreator.class);
 
     // not final so we can mutate for unit test
     private SerializationSchema<RowData> serializationSchema;
@@ -66,30 +67,6 @@ public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
     private final List<String> requestBodyFields;
     private final Map<String, String> requestUrlMap;
 
-    /**
-     * Construct a Generic JSON and URL query creator.
-     *
-     * @param httpMethod the requested http method
-     * @param serializationSchema    serialization schema for RowData
-     * @param requestQueryParamsFields query param fields
-     * @param requestBodyFields body fields used for PUT and POSTs
-     * @param requestUrlMap url map
-     * @param lookupRow     lookup row itself.
-     */
-    public GenericJsonAndUrlQueryCreator(final String httpMethod,
-                                         final SerializationSchema<RowData>
-                                             serializationSchema,
-                                         final List<String>  requestQueryParamsFields,
-                                         final List<String> requestBodyFields,
-                                         final Map<String, String> requestUrlMap,
-                                         final LookupRow lookupRow) {
-        this.httpMethod = httpMethod;
-        this.serializationSchema = serializationSchema;
-        this.lookupRow = lookupRow;
-        this.requestQueryParamsFields = requestQueryParamsFields;
-        this.requestBodyFields = requestBodyFields;
-        this.requestUrlMap = requestUrlMap;
-    }
     @VisibleForTesting
     void setSerializationSchema(SerializationSchema<RowData>
                                         serializationSchema) {
@@ -109,9 +86,7 @@ public LookupQueryInfo createLookupQuery(final RowData lookupDataRow) {
             jsonObject = (ObjectNode) ObjectMapperAdapter.instance().readTree(
                     serializationSchema.serialize(lookupDataRow));
         } catch (IOException e) {
-            String message = "Unable to parse the lookup arguments to json.";
-            log.error(message, e);
-            throw new RuntimeException(message, e);
+            throw new RuntimeException("Unable to parse the lookup arguments to json.", e);
         }
         // Parameters are encoded as query params for GET and none GET.
         // Later code will turn these query params into the body for PUTs and POSTs
@@ -132,17 +107,15 @@ public LookupQueryInfo createLookupQuery(final RowData lookupDataRow) {
                 lookupQuery = ObjectMapperAdapter.instance()
                         .writeValueAsString(jsonObject.retain(requestBodyFields));
             } catch (JsonProcessingException e) {
-                final String message = "Unable to convert Json Object to a string";
-                log.error(message, e);
-                throw new RuntimeException(message,e);
+                throw new RuntimeException("Unable to convert Json Object to a string", e);
             }
             // body parameters
             // use the request json object to scope the required fields and the lookupArgs as values
             bodyBasedUrlQueryParams = createBodyBasedParams(lookupArgs,
                     jsonObjectForQueryParams);
         }
         // add the path map
-        final Map<String, String> pathBasedUrlParams = createPathBasedParams(lookupArgs,
+        final Map<String, String> pathBasedUrlParams = createURLPathBasedParams(lookupArgs,
                 requestUrlMap);
 
         return new LookupQueryInfo(lookupQuery, bodyBasedUrlQueryParams, pathBasedUrlParams);
@@ -195,26 +168,25 @@ private Map<String, String> createBodyBasedParams(final Collection<LookupArg> ar
         return mapOfJsonKeyToLookupArg;
     }
     /**
-     * Create map of the json key to the lookup argument
-     * value. This is used for body based content.
+     * Create map of insert name to column name for path inserts
      * @param args lookup arguments
      * @param urlMap map of insert name to column name
      * @return map of field content to the lookup argument value.
      */
-    private Map<String, String> createPathBasedParams(final Collection<LookupArg> args,
-                                                     Map<String, String> urlMap ) {
-        Map<String, String> mapOfJsonKeyToLookupArg = new LinkedHashMap<>();
+    private Map<String, String> createURLPathBasedParams(final Collection<LookupArg> args,
+                                                         Map<String, String> urlMap ) {
+        Map<String, String> mapOfinsertKeyToLookupArg = new LinkedHashMap<>();
         if (urlMap != null) {
             for (String key: urlMap.keySet()) {
                 for (final LookupArg arg : args) {
                     if (arg.getArgName().equals(key)) {
-                        mapOfJsonKeyToLookupArg.put(
+                        mapOfinsertKeyToLookupArg.put(
                                 urlMap.get(key), arg.getArgValue());
                     }
                 }
             }
         }
-        return mapOfJsonKeyToLookupArg;
+        return mapOfinsertKeyToLookupArg;
     }
     /**
      * Convert json object to query params string
@@ -235,13 +207,13 @@ static String convertToQueryParameters(final ObjectNode jsonObject, String enc)
                 result.add(fieldName + "="
                         + URLEncoder.encode(fieldValue, enc));
             } catch (UnsupportedEncodingException e) {
-                final String message =
-                        "Failed to encode the value of the query parameter name "
-                                + fieldName
-                                + ": "
-                                + fieldValue;
-                log.error(message, e);
-                throw new RuntimeException(message, e);
+                throw new RuntimeException("Failed to encode the value of the query parameter name "
+                        + fieldName
+                        + ": "
+                        + fieldValue
+                        + " using encoding "
+                        + enc,
+                        e);
             }
         });
 
@@ -257,11 +229,8 @@ private void checkOpened() {
                                         GenericJsonAndUrlQueryCreator.class));
                 this.schemaOpened = true;
             } catch (final Exception e) {
-                final String message =
-                        "Failed to initialize serialization schema for "
-                                + GenericJsonAndUrlQueryCreator.class;
-                log.error(message, e);
-                throw new FlinkRuntimeException(message, e);
+                throw new FlinkRuntimeException("Failed to initialize serialization schema for "
+                        + GenericJsonAndUrlQueryCreator.class, e);
             }
         }
     }

src/main/java/com/getindata/connectors/http/internal/utils/ApiLookupConstants.java

@@ -20,7 +20,7 @@ public class CustomFormatFactory implements SerializationFormatFactory {
     public static final String REQUIRED_OPTION = "required-option-one";
 
     /**
-     * Consider removing this static only used for testing only
+     * Used for testing custom format.
      */
     static boolean requiredOptionsWereUsed = false;
 

src/test/java/com/getindata/connectors/http/internal/table/lookup/querycreators/CustomFormatFactory.java

@@ -20,6 +20,8 @@
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.types.Row;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -30,84 +32,67 @@
 import static com.getindata.connectors.http.internal.table.lookup.querycreators.QueryCreatorUtils.getTableContext;
 
 class GenericJsonAndUrlQueryCreatorTest {
-    enum Operation {
-        Get("GET"),
-        Put("PUT"),
-        Post("POST")
-        ;
-        public final String label;
-        Operation() {
-            this("GET");
-        }
 
-        Operation(String label) {
-            this.label = label;
-        }
-    }
-    @Test
-    public void createLookupQueryTestStrAllOps() {
-        for (Operation op: Operation.values()) {
-            String operation = op.label;
-            String key = "key1";
-            String value = "val1";
-            // for GET this is the minimum config
-            List<String> query_params = List.of(key);
-            // Path param ArgPath required a stringified json object. As we have PersonBean
-            // we can use that.
-            Map<String, String> url_params = Map.of(key, key);
-            LookupRow lookupRow = new LookupRow()
-                    .addLookupEntry(
-                            new RowDataSingleValueLookupSchemaEntry(
-                                    key,
-                                    RowData.createFieldGetter(
-                                            DataTypes.STRING().getLogicalType(), 0)
-                            ));
-            DataType dataType = row(List.of(
-                    DataTypes.FIELD(key, DataTypes.STRING())
-            ));
-            lookupRow.setLookupPhysicalRowDataType(dataType);
-            ResolvedSchema resolvedSchema = ResolvedSchema.of(Column.physical(key,
-                    DataTypes.STRING()));
-            Configuration config = new Configuration();
-            config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS,
+    @ParameterizedTest
+    @ValueSource(strings = {"GET", "PUT", "POST" })
+    public void createLookupQueryTestStrAllOps(String operation) {
+        String key = "key1";
+        String value = "val1";
+        // for GET this is the minimum config
+        List<String> query_params = List.of(key);
+        // Path param ArgPath required a stringified json object. As we have PersonBean
+        // we can use that.
+        Map<String, String> url_params = Map.of(key, key);
+        LookupRow lookupRow = new LookupRow()
+                .addLookupEntry(
+                        new RowDataSingleValueLookupSchemaEntry(
+                                key,
+                                RowData.createFieldGetter(
+                                        DataTypes.STRING().getLogicalType(), 0)
+                        ));
+        DataType dataType = row(List.of(
+                DataTypes.FIELD(key, DataTypes.STRING())
+        ));
+        lookupRow.setLookupPhysicalRowDataType(dataType);
+        ResolvedSchema resolvedSchema = ResolvedSchema.of(Column.physical(key,
+                DataTypes.STRING()));
+        Configuration config = new Configuration();
+        config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS,
+                query_params);
+        if (!operation.equals("GET")) {
+            // add the body content for PUT and POST
+            config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_BODY_FIELDS,
                     query_params);
-            if (!op.equals("GET")) {
-                // add the body content for PUT and POST
-                config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_BODY_FIELDS,
-                        query_params);
-            }
-            config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_URL_MAP, url_params);
-            config.setString(LOOKUP_METHOD, operation);
-            // GIVEN
-
-            GenericJsonAndUrlQueryCreator universalJsonQueryCreator =
-                    (GenericJsonAndUrlQueryCreator) new GenericJsonAndUrlQueryCreatorFactory()
-                            .createLookupQueryCreator(
-                                    config,
-                                    lookupRow,
-                                    getTableContext(config,
-                                            resolvedSchema)
-                            );
-            var row = new GenericRowData(1);
-            row.setField(0, StringData.fromString(value));
-            var createdQuery = universalJsonQueryCreator.createLookupQuery(row);
-            // THEN
-            if (operation.equals("GET")) {
-                assertThat(createdQuery.getBodyBasedUrlQueryParameters()).isEmpty();
-                assertThat(createdQuery.getLookupQuery()).isEqualTo(key + "=" + value);
-            } else {
-                assertThat(createdQuery
-                        .getBodyBasedUrlQueryParameters())
-                        .isEqualTo(key + "=" + value);
-                assertThat(createdQuery.getLookupQuery()).isEqualTo(
-                        "{\""
-                        + key
-                        + "\":\"" + value
-                        + "\"}");
-            }
-            assertThat(createdQuery.getPathBasedUrlParameters().size() == 1).isTrue();
-            assertThat(createdQuery.getPathBasedUrlParameters().get(key)).isEqualTo(value);
         }
+        config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_URL_MAP, url_params);
+        config.setString(LOOKUP_METHOD, operation);
+        GenericJsonAndUrlQueryCreator universalJsonQueryCreator =
+                (GenericJsonAndUrlQueryCreator) new GenericJsonAndUrlQueryCreatorFactory()
+                        .createLookupQueryCreator(
+                                config,
+                                lookupRow,
+                                getTableContext(config,
+                                        resolvedSchema)
+                        );
+        var row = new GenericRowData(1);
+        row.setField(0, StringData.fromString(value));
+        var createdQuery = universalJsonQueryCreator.createLookupQuery(row);
+        // THEN
+        if (operation.equals("GET")) {
+            assertThat(createdQuery.getBodyBasedUrlQueryParameters()).isEmpty();
+            assertThat(createdQuery.getLookupQuery()).isEqualTo(key + "=" + value);
+        } else {
+            assertThat(createdQuery
+                    .getBodyBasedUrlQueryParameters())
+                    .isEqualTo(key + "=" + value);
+            assertThat(createdQuery.getLookupQuery()).isEqualTo(
+                    "{\""
+                            + key
+                            + "\":\"" + value
+                            + "\"}");
+        }
+        assertThat(createdQuery.getPathBasedUrlParameters().size() == 1).isTrue();
+        assertThat(createdQuery.getPathBasedUrlParameters().get(key)).isEqualTo(value);
     }
     @Test
     public void createLookupQueryTest() {
@@ -167,7 +152,6 @@ public void failserializationOpenTest() {
         List<String> paths_config =List.of("key1");
         String operation = "GET";
         String key = "key1";
-        String value = "val1";
 
         LookupRow lookupRow = new LookupRow()
                 .addLookupEntry(
@@ -183,7 +167,6 @@ public void failserializationOpenTest() {
         Configuration config = new Configuration();
         config.set(GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS, paths_config);
         config.setString(LOOKUP_METHOD, operation);
-        // GIVEN
 
         lookupRow.setLookupPhysicalRowDataType(dataType);
 
@@ -213,15 +196,19 @@ public byte[] serialize(RowData element) {
         });
     }
     @Test void convertToQueryParametersUnsupportedEncodingTest() {
+        // GIVEN
         ObjectMapper mapper = ObjectMapperAdapter.instance();
         PersonBean person = new PersonBean("aaa", "bbb");
+        // WHEN
         JsonNode personNode = mapper.valueToTree(person);
+        // THEN
         assertThrows(RuntimeException.class, () -> {
             GenericJsonAndUrlQueryCreator.convertToQueryParameters(
                     (ObjectNode) personNode, "bad encoding");
         });
     }
     @Test void rowDataToRowTest() {
+        // GIVEN
         // String
         String KEY1 = "key1";
         String KEY2 = "key2";
@@ -238,7 +225,9 @@ public byte[] serialize(RowData element) {
                 DataTypes.FIELD(KEY2, DataTypes.DATE()),
                 DataTypes.FIELD(KEY3, DataTypes.TIMESTAMP_LTZ())
         ));
+        // WHEN
         Row row =  GenericJsonAndUrlQueryCreator.rowDataToRow(rowData, dataType);
+        // THEN
         assertThat(row.getField(KEY1).equals(VALUE));
         assertThat(row.getField(KEY2).equals("1970-01-01T00:00:00.010"));
         assertThat(row.getField(KEY3).equals("1970-01-01T00:00:00.010Z"));