Merge pull request #12 from getindata/ESP-98_SinkParameters

ESP-98_SinkParameters - pass headers via properties to Http client

Author: kristoffSC

.gitignore

@@ -6,3 +6,4 @@ target
 /src/main/main.iml
 /src/test/test.iml
 /flink-http-connector.iml
+/dependency-reduced-pom.xml

CHANGELOG.md

@@ -6,6 +6,7 @@
   Methods defined in classes located outside "internal" package are considered "public API".
   Any changes to those methods should be communicated as "not backward compatible" and should be avoided.
 - Add checkstyle configuration to "dev" folder. Add checkstyle check during maven build
+- Add HTTP sink client header configuration via properties.
 
 ## [0.2.0] - 2022-07-06
 

README.md

@@ -66,8 +66,8 @@ CREATE TABLE http (
   id bigint,
   some_field string
 ) WITH (
-  'connector' = 'http-sink'
-  'url' = 'http://example.com/myendpoint'
+  'connector' = 'http-sink',
+  'url' = 'http://example.com/myendpoint',
   'format' = 'json'
 )
 ```
@@ -82,6 +82,52 @@ Due to the fact that `HttpSink` sends bytes inside HTTP request's body, one can
 
 Other examples of usage of the Table API can be found in [some tests](src/test/java/com/getindata/connectors/http/table/HttpDynamicSinkInsertTest.java).
 
+#### Http headers (currently supported only for HTTP Sink)
+It is possible to set HTTP headers that will be added to HTTP request send by sink connector.
+Headers are defined via property key `gid.connector.http.sink.header.HEADER_NAME = header value` for example:
+`gid.connector.http.sink.header.X-Content-Type-Options = nosniff`.
+Properties can be set via Sink builder or Property object:
+```java
+HttpSink.<String>builder()
+      .setEndpointUrl("http://example.com/myendpoint")
+      .setElementConverter(
+          (s, _context) -> new HttpSinkRequestEntry("POST", s.getBytes(StandardCharsets.UTF_8)))
+      .setProperty("gid.connector.http.sink.header.X-Content-Type-Options", "nosniff")
+      .build();
+```
+or
+
+```java
+Properties properties = Properties();
+properties.setProperty("gid.connector.http.sink.header.X-Content-Type-Options", "nosniff");
+
+HttpSink.<String>builder()
+      .setEndpointUrl("http://example.com/myendpoint")
+      .setElementConverter(
+          (s, _context) -> new HttpSinkRequestEntry("POST", s.getBytes(StandardCharsets.UTF_8)))
+      .setProperties(properties)
+      .build();
+```
+
+In Table/SQL API, headers can be set using http sink table DDL. In example below, HTTP request done for `http` table will contain three headers:
+- `Origin`
+- `X-Content-Type-Options`
+- `Content-Type`
+
+```roomsql
+CREATE TABLE http (
+  id bigint,
+  some_field string
+) WITH (
+  'connector' = 'http-sink',
+  'url' = 'http://example.com/myendpoint',
+  'format' = 'json',
+  'gid.connector.http.sink.header.Origin' = '*',
+  'gid.connector.http.sink.header.X-Content-Type-Options' = 'nosniff',
+  'gid.connector.http.sink.header.Content-Type' = 'application/json'
+)
+```
+
 ## Implementation
 Implementation of an HTTP source connector is based on Flink's `TableFunction` and `AsyncTableFunction` classes.  
 To be more specific we are using a `LookupTableSource`. Unfortunately Flink's new unified source interface [2] cannot be used for this type of source.