HTTP-74 Add path parameter support

CHANGELOG.md

@@ -2,9 +2,17 @@
 
 ## [Unreleased]
 
+###  Changed
+
+-   Changed  [LookupQueryInfo](src/main/java/com/getindata/connectors/http/internal/table/lookup/LookupQueryInfo.java)
+    Any custom implementation of this interface that aims to provide path-based requests is able to provide
+    the lookup query url with parameters surrounded by curly brackets. For example the supplied
+    URL `http://service/{customerId}`, will result in the lookup parameter `customerId` value being used
+    in the url.
+
 ### Fixed
-   
-    Moved junit support to junit 5, allowing junits to be run against flink 1.17 and 1.18.
+
+-   Moved junit support to junit 5, allowing junits to be run against flink 1.17 and 1.18.
 
 ## [0.12.0] - 2024-03-22
 

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

@@ -16,7 +16,7 @@
 
 /**
  * Implementation of {@link HttpRequestFactory} for REST calls that sends their parameters using
- * request body.
+ * request body or in the path.
  */
 @Slf4j
 public class BodyBasedRequestFactory extends RequestFactoryBase {
@@ -43,7 +43,7 @@ public BodyBasedRequestFactory(
     @Override
     protected Builder setUpRequestMethod(LookupQueryInfo lookupQueryInfo) {
         return HttpRequest.newBuilder()
-            .uri(constructBodyBasedUri(lookupQueryInfo))
+            .uri(constructUri(lookupQueryInfo))
             .method(methodName, BodyPublishers.ofString(lookupQueryInfo.getLookupQuery()))
             .timeout(Duration.ofSeconds(this.httpRequestTimeOutSeconds));
     }
@@ -53,17 +53,19 @@ protected Logger getLogger() {
         return log;
     }
 
-    URI constructBodyBasedUri(LookupQueryInfo lookupQueryInfo) {
+    URI constructUri(LookupQueryInfo lookupQueryInfo) {
         StringBuilder resolvedUrl = new StringBuilder(baseUrl);
         if (lookupQueryInfo.hasBodyBasedUrlQueryParameters()) {
             resolvedUrl.append(baseUrl.contains("?") ? "&" : "?")
                        .append(lookupQueryInfo.getBodyBasedUrlQueryParameters());
         }
+        resolvedUrl = resolvePathParameters(lookupQueryInfo, resolvedUrl);
 
         try {
             return new URIBuilder(resolvedUrl.toString()).build();
         } catch (URISyntaxException e) {
             throw new RuntimeException(e);
         }
     }
+
 }

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

@@ -60,7 +60,7 @@ URI constructGetUri(LookupQueryInfo lookupQueryInfo) {
             resolvedUrl.append(baseUrl.contains("?") ? "&" : "?")
                        .append(lookupQueryInfo.getLookupQuery());
         }
-
+        resolvedUrl = resolvePathParameters(lookupQueryInfo, resolvedUrl);
         try {
             return new URIBuilder(resolvedUrl.toString()).build();
         } catch (URISyntaxException e) {

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

@@ -12,6 +12,7 @@
 import com.getindata.connectors.http.internal.utils.uri.NameValuePair;
 import com.getindata.connectors.http.internal.utils.uri.URLEncodedUtils;
 
+
 /**
  * Holds the lookup query for an HTTP request.
  * The {@code  lookupQuery} either contain the query parameters for a GET operation
@@ -26,32 +27,47 @@ public class LookupQueryInfo implements Serializable {
 
     private final Map<String, String> bodyBasedUrlQueryParams;
 
+    private final Map<String, String> pathBasedUrlParams;
+
     public LookupQueryInfo(String lookupQuery) {
-        this(lookupQuery, null);
+        this(lookupQuery, null, null);
     }
 
-    public LookupQueryInfo(String lookupQuery, Map<String, String> bodyBasedUrlQueryParams) {
+    public LookupQueryInfo(String lookupQuery, Map<String, String> bodyBasedUrlQueryParams,
+                           Map<String, String> pathBasedUrlParams) {
         this.lookupQuery =
                 lookupQuery == null ? "" : lookupQuery;
         this.bodyBasedUrlQueryParams =
                 bodyBasedUrlQueryParams == null ? Collections.emptyMap() : bodyBasedUrlQueryParams;
+        this.pathBasedUrlParams =
+                pathBasedUrlParams == null ? Collections.emptyMap() : pathBasedUrlParams;
     }
 
     public String getBodyBasedUrlQueryParameters() {
         return URLEncodedUtils.format(
                 bodyBasedUrlQueryParams
                         .entrySet()
                         .stream()
+                        // sort the map by key to ensure there is a reliable order for unit tests
+                        .sorted(Map.Entry.comparingByKey())
                         .map(entry -> new NameValuePair(entry.getKey(), entry.getValue()))
                         .collect(Collectors.toList()),
                 StandardCharsets.UTF_8);
     }
 
+    public Map<String, String> getPathBasedUrlParameters() {
+        return pathBasedUrlParams;
+    }
+
     public boolean hasLookupQuery() {
         return !lookupQuery.isBlank();
     }
+
     public boolean hasBodyBasedUrlQueryParameters() {
         return !bodyBasedUrlQueryParams.isEmpty();
     }
 
+    public boolean hasPathBasedUrlParameters() {
+        return !pathBasedUrlParams.isEmpty();
+    }
 }

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

@@ -3,9 +3,11 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpRequest.Builder;
 import java.util.Arrays;
+import java.util.Map;
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.FlinkRuntimeException;
 import org.slf4j.Logger;
 
 import com.getindata.connectors.http.LookupQueryCreator;
@@ -82,6 +84,24 @@ public HttpLookupSourceRequestEntry buildLookupRequest(RowData lookupRow) {
      */
     protected abstract Builder setUpRequestMethod(LookupQueryInfo lookupQuery);
 
+    protected static StringBuilder resolvePathParameters(LookupQueryInfo lookupQueryInfo,
+                                                         StringBuilder resolvedUrl) {
+        if (lookupQueryInfo.hasPathBasedUrlParameters()) {
+            for (Map.Entry<String, String> entry :
+                    lookupQueryInfo.getPathBasedUrlParameters().entrySet()) {
+                String pathParam = "{" + entry.getKey() + "}";
+                int startIndex = resolvedUrl.indexOf(pathParam);
+                if (startIndex == -1) {
+                    throw new FlinkRuntimeException(
+                            "Unexpected error while parsing the URL for path parameters.");
+                }
+                int endIndex = startIndex + pathParam.length();
+                resolvedUrl = resolvedUrl.replace(startIndex, endIndex, entry.getValue());
+            }
+        }
+        return resolvedUrl;
+    }
+
     @VisibleForTesting
     String[] getHeadersAndValues() {
         return Arrays.copyOf(headersAndValues, headersAndValues.length);

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

@@ -0,0 +1,124 @@
+package com.getindata.connectors.http.internal.table.lookup;
+
+import java.net.URI;
+import java.util.Collection;
+import java.util.Map;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class BodyBasedRequestFactoryTest {
+
+    @ParameterizedTest
+    @MethodSource("configProvider")
+    void testconstructUri(TestSpec testSpec) throws Exception {
+        LookupQueryInfo lookupQueryInfo = new LookupQueryInfo(testSpec.url,
+                testSpec.bodyBasedUrlQueryParams,
+                testSpec.pathBasedUrlParams);
+        HttpLookupConfig httpLookupConfig = HttpLookupConfig.builder()
+                .lookupMethod(testSpec.lookupMethod)
+                .url(testSpec.url)
+                .useAsync(false)
+                .build();
+        BodyBasedRequestFactory bodyBasedRequestFactory =
+                new BodyBasedRequestFactory("test", null, null, httpLookupConfig);
+
+        URI uri = bodyBasedRequestFactory.constructUri(lookupQueryInfo);
+        assertThat(uri.toString()).isEqualTo(testSpec.expected);
+    }
+
+    private static class TestSpec {
+
+        Map<String, String> bodyBasedUrlQueryParams;
+        Map<String, String> pathBasedUrlParams;
+        String url;
+        String lookupMethod;
+        String expected;
+
+        private TestSpec(Map<String, String> bodyBasedUrlQueryParams,
+                         Map<String, String> pathBasedUrlParams,
+                         String url,
+                         String lookupMethod,
+                         String expected) {
+            this.bodyBasedUrlQueryParams = bodyBasedUrlQueryParams;
+            this.pathBasedUrlParams = pathBasedUrlParams;
+            this.url = url;
+            this.lookupMethod = lookupMethod;
+            this.expected = expected;
+        }
+
+        @Override
+        public String toString() {
+            return "TestSpec{"
+                    + "bodyBasedUrlQueryParams="
+                    + bodyBasedUrlQueryParams
+                    + ", pathBasedUrlParams="
+                    + pathBasedUrlParams
+                    + ", url="
+                    + url
+                    + ", lookupMethod="
+                    + lookupMethod
+                    + ", expected="
+                    + expected
+                    + '}';
+        }
+    }
+
+    static Collection<TestSpec> configProvider() {
+        return ImmutableList.<TestSpec>builder()
+                .addAll(getTestSpecs("GET"))
+                .addAll(getTestSpecs("POST"))
+                .build();
+    }
+
+    @NotNull
+    private static ImmutableList<TestSpec> getTestSpecs(String lookupMethod) {
+        return ImmutableList.of(
+                // 1 path param
+                new TestSpec(
+                        null,
+                        Map. of("param1", "value1"),
+                        "http://service/{param1}",
+                        lookupMethod,
+                        "http://service/value1"),
+                // 2 path param
+                new TestSpec(
+                        null,
+                        Map. of("param1", "value1", "param2", "value2"),
+                        "http://service/{param1}/param2/{param2}",
+                        lookupMethod,
+                        "http://service/value1/param2/value2"),
+                // 1 query param
+                new TestSpec(
+                        Map. of("param3", "value3"),
+                        null,
+                        "http://service",
+                        lookupMethod,
+                        "http://service?param3=value3"),
+                // 1 query param with a parameter on base url
+                new TestSpec(
+                        Map. of("param3", "value3"),
+                        null,
+                        "http://service?extrakey=extravalue",
+                        lookupMethod,
+                        "http://service?extrakey=extravalue&param3=value3"),
+                // 2 query params
+                new TestSpec(
+                        Map. of("param3", "value3", "param4", "value4"),
+                        null,
+                        "http://service",
+                        lookupMethod,
+                        "http://service?param3=value3&param4=value4"),
+                // 2 query params and 2 path params
+                new TestSpec(
+                        Map. of("param3", "value3", "param4", "value4"),
+                        Map. of("param1", "value1", "param2", "value2"),
+                        "http://service/{param1}/param2/{param2}",
+                        lookupMethod,
+                        "http://service/value1/param2/value2?param3=value3&param4=value4")
+        );
+    }
+}

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

@@ -165,7 +165,8 @@ public void shouldBuildBodyBasedClientUri() {
         urlBodyBasedQueryParameters.put("key1", "value1");
         urlBodyBasedQueryParameters.put("key2", "value2");
 
-        LookupQueryInfo lookupQueryInfo = new LookupQueryInfo("{}", urlBodyBasedQueryParameters);
+        LookupQueryInfo lookupQueryInfo = new LookupQueryInfo("{}",
+                urlBodyBasedQueryParameters, null);
 
         // WHEN
         HttpRequest httpRequest = requestFactory.setUpRequestMethod(lookupQueryInfo).build();

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

@@ -14,7 +14,7 @@ public void testConfiguredLookupQuery() {
         String lookupQuery = "{\"param1\": \"value1\"}";
         Map<String, String> bodyBasedUrlQueryParameters = Map.of("key1", "value1");
 
-        lookupQueryInfo = new LookupQueryInfo(lookupQuery, bodyBasedUrlQueryParameters);
+        lookupQueryInfo = new LookupQueryInfo(lookupQuery, bodyBasedUrlQueryParameters, null);
 
         assertThat(lookupQueryInfo.hasLookupQuery()).isTrue();
         assertThat(lookupQueryInfo.getLookupQuery()).isEqualTo(lookupQuery);
@@ -23,7 +23,7 @@ public void testConfiguredLookupQuery() {
     }
     @Test
     public void testEmptyLookupQueryInfo() {
-        lookupQueryInfo = new LookupQueryInfo(null, null);
+        lookupQueryInfo = new LookupQueryInfo(null, null, null);
 
         assertThat(lookupQueryInfo.hasLookupQuery()).isFalse();
         assertThat(lookupQueryInfo.hasBodyBasedUrlQueryParameters()).isFalse();
@@ -32,13 +32,57 @@ public void testEmptyLookupQueryInfo() {
     }
 
     @Test
-    public void testBodyBasedUrlQueryParams() {
+    public void test1BodyParam() {
         Map<String, String> bodyBasedUrlQueryParameters = Map.of("key1", "value1");
 
-        lookupQueryInfo = new LookupQueryInfo(null, bodyBasedUrlQueryParameters);
+        lookupQueryInfo = new LookupQueryInfo(null, bodyBasedUrlQueryParameters, null);
 
         assertThat(lookupQueryInfo.hasLookupQuery()).isFalse();
         assertThat(lookupQueryInfo.hasBodyBasedUrlQueryParameters()).isTrue();
         assertThat(lookupQueryInfo.getBodyBasedUrlQueryParameters()).isEqualTo("key1=value1");
     }
+
+    @Test
+    public void test1PathParam() {
+        Map<String, String> pathBasedUrlPathParameters = Map.of("key1", "value1");
+
+        lookupQueryInfo = new LookupQueryInfo("http://service/{key1}",
+                null, pathBasedUrlPathParameters);
+
+        assertThat(lookupQueryInfo.hasLookupQuery()).isTrue();
+        assertThat(lookupQueryInfo.hasPathBasedUrlParameters()).isTrue();
+        assertThat(lookupQueryInfo.getPathBasedUrlParameters())
+                .isEqualTo(pathBasedUrlPathParameters);
+    }
+    @Test
+    public void test2Path2BodyParams() {
+        Map<String, String> pathBasedUrlPathParameters =
+                Map.of("key1", "value1", "key2", "value2");
+        Map<String, String> bodyBasedQueryParameters =
+                Map.of("key3", "value3", "key4", "value4");
+
+        lookupQueryInfo = new LookupQueryInfo(null,
+                bodyBasedQueryParameters, pathBasedUrlPathParameters);
+
+        assertThat(lookupQueryInfo.hasLookupQuery()).isFalse();
+        assertThat(lookupQueryInfo.hasPathBasedUrlParameters()).isTrue();
+        assertThat(lookupQueryInfo.getPathBasedUrlParameters())
+                .isEqualTo(pathBasedUrlPathParameters);
+        assertThat(lookupQueryInfo.hasBodyBasedUrlQueryParameters())
+                .isTrue();
+        assertThat(lookupQueryInfo.getBodyBasedUrlQueryParameters())
+                .isEqualTo("key3=value3&key4=value4");
+    }
+
+    @Test
+    public void test2PathParams() {
+        Map<String, String> pathBasedUrlPathParameters = Map.of("key1", "value1", "key2", "value2");
+
+        lookupQueryInfo = new LookupQueryInfo(null, null, pathBasedUrlPathParameters);
+
+        assertThat(lookupQueryInfo.hasLookupQuery()).isFalse();
+        assertThat(lookupQueryInfo.hasPathBasedUrlParameters()).isTrue();
+        assertThat(lookupQueryInfo.getPathBasedUrlParameters())
+                .isEqualTo(pathBasedUrlPathParameters);
+    }
 }