http-99 rebase

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

Author: davidradl

CHANGELOG.md

@@ -2,12 +2,8 @@
 
 ## [Unreleased]
 
-<<<<<<< HEAD
 -   Added support for generic json and URL query creator 
 
-=======
--   Added support for generic json and URL query creator
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
 -   Retries support for source table:
     -   Auto retry on IOException and user-defined http codes - parameter `gid.connector.http.source.lookup.retry-codes`.
     -   Parameters `gid.connector.http.source.lookup.error.code.exclude"` and `gid.connector.http.source.lookup.error.code` were replaced by `gid.connector.http.source.lookup.ignored-response-codes`.

README.md

@@ -1,10 +1,7 @@
-<<<<<<< HEAD
-=======
 /*
  * © Copyright IBM Corp. 2025
  */
 
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
 package com.getindata.connectors.http.internal.table.lookup.querycreators;
 
 import java.io.IOException;
@@ -13,21 +10,12 @@
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 
-<<<<<<< HEAD
-import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
-=======
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
 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;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
-<<<<<<< HEAD
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.Preconditions;
-=======
 import org.apache.flink.table.api.DataTypes.Field;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
@@ -36,9 +24,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;
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
 
 import com.getindata.connectors.http.LookupArg;
 import com.getindata.connectors.http.LookupQueryCreator;
@@ -47,18 +32,6 @@
 import com.getindata.connectors.http.internal.utils.SerializationSchemaUtils;
 
 /**
-<<<<<<< HEAD
- * <p>Generic JSON and URL query creator; in addition to be able to map columns to json requests,
- * it allows url inserts to be mapped to column names using templating.</p>
- * <p>For GETs, column names are mapped to query parameters. e.g. for
- * <code>GenericJsonAndUrlQueryCreator.REQUEST_PARAM_FIELDS</code> = "id1;id2"
- * and url of http://base. At lookup time with values of id1=1 and id2=2 a call of
- * http/base?id1=1&amp;id2=2 will be issued.</p>
- * <p>For PUT and POST, parameters are mapped to the json body  e.g. for
- * REQUEST_PARAM_FIELDS = "id1;id2" and url of http://base. At lookup time with values of id1=1 and
- * id2=2 as call of http/base will be issued with a json payload of {"id1":1,"id2":2}</p>
- * <p>For all http methods, url segments can be used to include lookup up values. Using the map from
-=======
  * Generic JSON and URL query creator; in addition to be able to map columns to json requests,
  * it allows url inserts to be mapped to column names using templating.
  * <br>
@@ -72,33 +45,16 @@
  * id2=2 as call of http/base will be issued with a json payload of {"id1":1,"id2":2}
  * <br>
  * For all http methods, url segments can be used to include lookup up values. Using the map from
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
  * <code>GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP</code> which has a key of the insert and the
  * value of the associated column.
  * e.g. for <code>GenericJsonAndUrlQueryCreator.REQUEST_URL_MAP</code> = "key1":"col1"
  * and url of http://base/{key1}. At lookup time with values of col1="aaaa" a call of
-<<<<<<< HEAD
- * http/base/aaaa will be issued.</p>
- */
-@Slf4j
-@Builder
-public class GenericJsonAndUrlQueryCreator implements LookupQueryCreator {
-    private static final long serialVersionUID = 1L;
-    private final String httpMethod;
-    // not final so we can mutate for unit test
-    private SerializationSchema<RowData> serializationSchema;
-    private final List<String>  requestQueryParamsFields;
-    private boolean schemaOpened = false;
-    private final List<String> requestBodyFields;
-    private final Map<String, String> requestUrlMap;
-    private final LookupRow lookupRow;
-=======
  * http/base/aaaa will be issued.
  *
  */
+@Slf4j
 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;
@@ -133,7 +89,6 @@ public GenericJsonAndUrlQueryCreator(final String httpMethod,
         this.requestBodyFields = requestBodyFields;
         this.requestUrlMap = requestUrlMap;
     }
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
     @VisibleForTesting
     void setSerializationSchema(SerializationSchema<RowData>
                                         serializationSchema) {
@@ -153,13 +108,9 @@ public LookupQueryInfo createLookupQuery(final RowData lookupDataRow) {
             jsonObject = (ObjectNode) ObjectMapperAdapter.instance().readTree(
                     serializationSchema.serialize(lookupDataRow));
         } catch (IOException e) {
-<<<<<<< HEAD
-            throw new RuntimeException("Unable to parse the lookup arguments to json.", e);
-=======
             String message = "Unable to parse the lookup arguments to json.";
             log.error(message, e);
             throw new RuntimeException(message, e);
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
         }
         // 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
@@ -180,34 +131,24 @@ public LookupQueryInfo createLookupQuery(final RowData lookupDataRow) {
                 lookupQuery = ObjectMapperAdapter.instance()
                         .writeValueAsString(jsonObject.retain(requestBodyFields));
             } catch (JsonProcessingException e) {
-<<<<<<< HEAD
-                throw new RuntimeException("Unable to convert Json Object to a string", e);
-=======
                 final String message = "Unable to convert Json Object to a string";
                 log.error(message, e);
                 throw new RuntimeException(message,e);
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
             }
             // 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
-<<<<<<< HEAD
-        final Map<String, String> pathBasedUrlParams = createURLPathBasedParams(lookupArgs,
-=======
         final Map<String, String> pathBasedUrlParams = createPathBasedParams(lookupArgs,
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
                 requestUrlMap);
 
         return new LookupQueryInfo(lookupQuery, bodyBasedUrlQueryParams, pathBasedUrlParams);
 
     }
 
     /**
-<<<<<<< HEAD
-=======
      * Create a Row from a RowData and DataType
      * @param lookupRowData the lookup RowData
      * @param rowType the datatype
@@ -230,19 +171,14 @@ static Row rowDataToRow(final RowData lookupRowData, final DataType rowType) {
     }
 
     /**
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
      * Create map of the json key to the lookup argument
      * value. This is used for body based content.
      * @param args lookup arguments
      * @param objectNode object node
      * @return map of field content to the lookup argument value.
      */
     private Map<String, String> createBodyBasedParams(final Collection<LookupArg> args,
-<<<<<<< HEAD
-                                                      ObjectNode objectNode ) {
-=======
                                                               ObjectNode objectNode ) {
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
         Map<String, String> mapOfJsonKeyToLookupArg = new LinkedHashMap<>();
         Iterator<Map.Entry<String, JsonNode>> iterator = objectNode.fields();
         iterator.forEachRemaining(field -> {
@@ -258,45 +194,26 @@ private Map<String, String> createBodyBasedParams(final Collection<LookupArg> ar
         return mapOfJsonKeyToLookupArg;
     }
     /**
-<<<<<<< HEAD
-     * Create map of insert name to column name for path inserts
-=======
      * Create map of the json key to the lookup argument
      * value. This is used for body based content.
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
      * @param args lookup arguments
      * @param urlMap map of insert name to column name
      * @return map of field content to the lookup argument value.
      */
-<<<<<<< HEAD
-    private Map<String, String> createURLPathBasedParams(final Collection<LookupArg> args,
-                                                         Map<String,
-                                                         String> urlMap ) {
-        Map<String, String> mapOfinsertKeyToLookupArg = new LinkedHashMap<>();
-=======
     private Map<String, String> createPathBasedParams(final Collection<LookupArg> args,
                                                      Map<String, String> urlMap ) {
         Map<String, String> mapOfJsonKeyToLookupArg = new LinkedHashMap<>();
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
         if (urlMap != null) {
             for (String key: urlMap.keySet()) {
                 for (final LookupArg arg : args) {
                     if (arg.getArgName().equals(key)) {
-<<<<<<< HEAD
-                        mapOfinsertKeyToLookupArg.put(
-=======
                         mapOfJsonKeyToLookupArg.put(
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
                                 urlMap.get(key), arg.getArgValue());
                     }
                 }
             }
         }
-<<<<<<< HEAD
-        return mapOfinsertKeyToLookupArg;
-=======
         return mapOfJsonKeyToLookupArg;
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
     }
     /**
      * Convert json object to query params string
@@ -317,23 +234,13 @@ static String convertToQueryParameters(final ObjectNode jsonObject, String enc)
                 result.add(fieldName + "="
                         + URLEncoder.encode(fieldValue, enc));
             } catch (UnsupportedEncodingException e) {
-<<<<<<< HEAD
-                throw new RuntimeException("Failed to encode the value of the query parameter name "
-                        + fieldName
-                        + ": "
-                        + fieldValue
-                        + " using encoding "
-                        + enc,
-                        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);
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
             }
         });
 
@@ -349,16 +256,11 @@ private void checkOpened() {
                                         GenericJsonAndUrlQueryCreator.class));
                 this.schemaOpened = true;
             } catch (final Exception e) {
-<<<<<<< HEAD
-                throw new FlinkRuntimeException("Failed to initialize serialization schema for "
-                        + GenericJsonAndUrlQueryCreator.class, e);
-=======
                 final String message =
                         "Failed to initialize serialization schema for "
                                 + GenericJsonAndUrlQueryCreator.class;
                 log.error(message, e);
                 throw new FlinkRuntimeException(message, e);
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
             }
         }
     }

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

@@ -16,8 +16,6 @@
 import org.apache.flink.table.factories.DynamicTableFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.factories.SerializationFormatFactory;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import static org.apache.flink.configuration.ConfigOptions.key;
 
 import com.getindata.connectors.http.LookupQueryCreator;
@@ -39,11 +37,6 @@
 @SuppressWarnings({"checkstyle:RegexpSingleline", "checkstyle:LineLength"})
 public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorFactory {
     private static final long serialVersionUID = 1L;
-<<<<<<< HEAD
-    private static final Logger log =
-            LogManager.getLogger(GenericJsonAndUrlQueryCreatorFactory.class);
-=======
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
 
     public static final String ID = "generic-json-url";
 
@@ -73,11 +66,7 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorF
                                  + "as url segments. Parses a string as a map of strings. "
                                  + "<br>"
                                  + "For example if there are table columns called customerId"
-<<<<<<< HEAD
-                                 + " and orderId, then specifying value customerId:cid,orderID:oid"
-=======
                                  + " and orderId, then specifying value customerId:cid1,orderID:oid"
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
                                  + " and a url of https://myendpoint/customers/{cid}/orders/{oid}"
                                  + " will mean that the url used for the lookup query will"
                                  + " dynamically pickup the values for customerId, orderId"
@@ -125,16 +114,6 @@ public LookupQueryCreator createLookupQueryCreator(final ReadableConfig readable
                     encoder.createRuntimeEncoder(null,
                             lookupRow.getLookupPhysicalRowDataType());
         }
-<<<<<<< HEAD
-        return GenericJsonAndUrlQueryCreator.builder()
-                .httpMethod(httpMethod)
-                .serializationSchema(jsonSerializationSchema)
-                .requestQueryParamsFields(requestQueryParamsFields)
-                .requestBodyFields(requestBodyFields)
-                .requestUrlMap(requestUrlMap)
-                .lookupRow(lookupRow)
-                .build();
-=======
         // create using config parameter values and specify serialization
         // schema from json format.
         return new GenericJsonAndUrlQueryCreator(httpMethod,
@@ -143,7 +122,6 @@ public LookupQueryCreator createLookupQueryCreator(final ReadableConfig readable
                 requestBodyFields,
                 requestUrlMap,
                 lookupRow);
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
     }
 
     @Override

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

@@ -20,11 +20,7 @@ public class CustomFormatFactory implements SerializationFormatFactory {
     public static final String REQUIRED_OPTION = "required-option-one";
 
     /**
-<<<<<<< HEAD
-     * Used for testing custom format.
-=======
      * Consider removing this static only used for testing only
->>>>>>> 6c68722 (HTTP-99 Generic Json url query creator)
      */
     static boolean requiredOptionsWereUsed = false;