ESP-98ESP-98_SinkParameters - add properties to Table API.

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Author: Krzysztof Chmielewski

src/main/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicTableSinkFactory.java

@@ -11,6 +11,7 @@
 import org.apache.flink.table.connector.sink.DynamicTableSink;
 import org.apache.flink.table.factories.FactoryUtil;
 
+import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.utils.ConfigUtils;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.INSERT_METHOD;
 import static com.getindata.connectors.http.internal.table.sink.HttpDynamicSinkConnectorOptions.URL;
@@ -33,7 +34,8 @@ public DynamicTableSink createDynamicTableSink(Context context) {
         ReadableConfig tableOptions = factoryContext.getTableOptions();
 
         // Validate configuration
-        FactoryUtil.createTableFactoryHelper(this, context).validate();
+        FactoryUtil.createTableFactoryHelper(this, context)
+            .validateExcept(HttpConnectorConfigConstants.GID_CONNECTOR_HTTP);
         validateHttpSinkOptions(tableOptions);
         Properties asyncSinkProperties =
             new AsyncSinkConfigurationValidator(tableOptions).getValidatedConfigurations();

src/test/java/com/getindata/connectors/http/internal/table/sink/HttpDynamicSinkInsertTest.java

@@ -2,6 +2,7 @@
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 
 import com.github.tomakehurst.wiremock.WireMockServer;
 import com.github.tomakehurst.wiremock.http.RequestMethod;
@@ -162,4 +163,48 @@ public void testHttpDynamicSinkRawFormat() throws Exception {
         assertEquals(RequestMethod.POST, request.getMethod());
         assertEquals("application/octet-stream", request.getHeader("Content-Type"));
     }
+
+    @Test
+    public void testHttpRequestWithHeadersFromDdl()
+        throws ExecutionException, InterruptedException {
+        String originHeaderValue = "*";
+        String xContentTypeOptionsHeaderValue = "nosniff";
+        String contentTypeHeaderValue = "application/json";
+
+        wireMockServer.stubFor(any(urlPathEqualTo("/myendpoint")).willReturn(ok()));
+
+        final String createTable =
+            String.format(
+                "CREATE TABLE http (\n"
+                    + "  last_name string"
+                    + ") with (\n"
+                    + "  'connector' = '%s',\n"
+                    + "  'url' = '%s',\n"
+                    + "  'format' = 'raw',\n"
+                    + "  'gid.connector.http.sink.header.Origin' = '%s',\n"
+                    + "  'gid.connector.http.sink.header.X-Content-Type-Options' = '%s',\n"
+                    + "  'gid.connector.http.sink.header.Content-Type' = '%s'\n"
+                    + ")",
+                HttpDynamicTableSinkFactory.IDENTIFIER,
+                "http://localhost:" + SERVER_PORT + "/myendpoint",
+                originHeaderValue,
+                xContentTypeOptionsHeaderValue,
+                contentTypeHeaderValue
+            );
+
+        tEnv.executeSql(createTable);
+
+        final String insert = "INSERT INTO http VALUES ('Clee')";
+        tEnv.executeSql(insert).await();
+
+        var postedRequests = wireMockServer.findAll(anyRequestedFor(urlPathEqualTo("/myendpoint")));
+        assertEquals(1, postedRequests.size());
+
+        var request = postedRequests.get(0);
+        assertEquals("Clee", request.getBodyAsString());
+        assertEquals(RequestMethod.POST, request.getMethod());
+        assertEquals(contentTypeHeaderValue, request.getHeader("Content-Type"));
+        assertEquals(originHeaderValue, request.getHeader("Origin"));
+        assertEquals(xContentTypeOptionsHeaderValue, request.getHeader("X-Content-Type-Options"));
+    }
 }