diff --git a/docs/content.zh/docs/connectors/table/http.md b/docs/content.zh/docs/connectors/table/http.md index e73ada33..d1ec6d80 100644 --- a/docs/content.zh/docs/connectors/table/http.md +++ b/docs/content.zh/docs/connectors/table/http.md @@ -45,8 +45,8 @@ The HTTP source connector supports [Lookup Joins](https://nightlies.apache.org/f * [Using a HTTP Lookup Source in a lookup join](#using-a-http-lookup-source-in-a-lookup-join) * [Lookup Source Connector Options](#lookup-source-connector-options) * [Query Creators](#query-creators) - * [generic-json-url Query Creator](#generic-json-url-query-creator) - * [generic-json-url Query Creator](#generic-json-url-query-creator-1) + * [http-generic-json-url Query Creator](#http-generic-json-url-query-creator) + * [http-generic-json-url Query Creator](#http-generic-json-url-query-creator-1) * [Http headers](#http-headers) * [Timeouts](#timeouts) * [Source table HTTP status code](#source-table-http-status-code) @@ -81,8 +81,11 @@ The GetInData HTTP connector was donated to Flink in [FLIP-532](https://cwiki.ap The Flink connector does have some changes that you need to be aware of if you are migrating from using the original connector: * Existing java applications will need to be recompiled to pick up the new flink package names. -* Existing application and SQL need to be amended to use the new connector option names. The new option names do not have +* Existing application and SQL need to be amended to use the new connector option names. The new option names do not have the _com.getindata.http_ prefix, the prefix is now _http_. +* The name of the connector and the identifiers of components that are discovered have been changed, so that the GetInData jar file can co-exist +* with this connector's jar file. Be aware that if you have created custom pluggable components; you will need to recompile against +* this connector. ## Working with HTTP lookup source tables @@ -102,7 +105,7 @@ CREATE TABLE Customers ( > > ) WITH ( -'connector' = 'rest-lookup', +'connector' = 'http', 'format' = 'json', 'url' = 'http://localhost:8080/client', 'asyncPolling' = 'true' @@ -157,7 +160,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options, | Option | Required | Description/Value | |:-----------------------------------------------------------------------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| connector | required | The Value should be set to _rest-lookup_ | +| connector | required | The Value should be set to _http_ | | format | required | Flink's format name that should be used to decode REST response, Use `json` for a typical REST endpoint. | | url | required | The base URL that should be use for GET requests. For example _http://localhost:8080/client_ | | asyncPolling | optional | true/false - determines whether Async Polling should be used. Mechanism is based on Flink's Async I/O. | @@ -218,13 +221,13 @@ The HTTP connector supplies a number of Query Creators that you can use define t -
generic-json-url
+
http-generic-json-url
✓ ✓ ✓ -
generic-get-query
+
http-generic-get-query
<✓ for GETs/td> ✓ for PUTs and POSTs @@ -233,20 +236,20 @@ The HTTP connector supplies a number of Query Creators that you can use define t -### generic-json-url Query Creator +### http-generic-json-url Query Creator -The recommended Query creator for json is called _generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports +The recommended Query creator for json is called _http-generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports POST, PUT and GET operations. This query creator allows you to issue json requests without needing to code your own custom http connector. The mappings from columns to the json request are supplied in the query creator configuration parameters `http.request.query-param-fields`, `http.request.body-fields` and `http.request.url-map`. -### generic-json-url Query Creator +### http-generic-json-url Query Creator -The default Query Creator is called _generic-json-url_. For body based queries such as POST/PUT requests, the +The default Query Creator is called _http-generic-json-url_. For body based queries such as POST/PUT requests, the ([GenericGetQueryCreator](flink-connector-http/src/main/java/org/apache/flink/connectors/http/internal/table/lookup/querycreators/GenericGetQueryCreator.java))is provided as a default query creator. This implementation uses Flink's [json-format](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/) to convert RowData object into Json String. For GET requests can be used for query parameter based queries. -The _generic-json-url_ allows for using custom formats that will perform serialization to Json. Thanks to this, users can create their own logic for converting RowData to Json Strings suitable for their HTTP endpoints and use this logic as custom format +The _http-generic-json-url_ allows for using custom formats that will perform serialization to Json. Thanks to this, users can create their own logic for converting RowData to Json Strings suitable for their HTTP endpoints and use this logic as custom format with HTTP Lookup connector and SQL queries. To create a custom format user has to implement Flink's `SerializationSchema` and `SerializationFormatFactory` interfaces and register custom format factory along other factories in `resources/META-INF.services/org.apache.flink.table.factories.Factory` file. This is common Flink mechanism for providing custom implementations for various factories. @@ -266,7 +269,7 @@ CREATE TABLE http-lookup ( id bigint, some_field string ) WITH ( - 'connector' = 'rest-lookup', + 'connector' = 'http', 'format' = 'json', 'url' = 'http://localhost:8080/client', 'asyncPolling' = 'true', diff --git a/docs/content/docs/connectors/table/http.md b/docs/content/docs/connectors/table/http.md index a2cc2339..ada26362 100644 --- a/docs/content/docs/connectors/table/http.md +++ b/docs/content/docs/connectors/table/http.md @@ -45,8 +45,8 @@ The HTTP source connector supports [Lookup Joins](https://nightlies.apache.org/f * [Using a HTTP Lookup Source in a lookup join](#using-a-http-lookup-source-in-a-lookup-join) * [Lookup Source Connector Options](#lookup-source-connector-options) * [Query Creators](#query-creators) - * [generic-json-url Query Creator](#generic-json-url-query-creator) - * [generic-json-url Query Creator](#generic-json-url-query-creator-1) + * [http-generic-json-url Query Creator](#http-generic-json-url-query-creator) + * [http-generic-json-url Query Creator](#http-generic-json-url-query-creator-1) * [Http headers](#http-headers) * [Timeouts](#timeouts) * [Source table HTTP status code](#source-table-http-status-code) @@ -81,8 +81,10 @@ The GetInData HTTP connector was donated to Flink in [FLIP-532](https://cwiki.ap The Flink connector does have some changes that you need to be aware of if you are migrating from using the original connector: * Existing java applications will need to be recompiled to pick up the new flink package names. -* Existing application and SQL need to be amended to use the new connector option names. The new option names do not have +* Existing application and SQL need to be amended to use the new connector option names. The new option names do not have the _com.getindata.http_ prefix, the prefix is now _http_. +* The name of the connector and the identifiers of components that are discovered have been changed, so that the GetInData jar file can co-exist +* with this connector's jar file. Be aware that if you have created custom pluggable components; you will need to recompile against ## Working with HTTP lookup source tables @@ -102,7 +104,7 @@ CREATE TABLE Customers ( > > ) WITH ( -'connector' = 'rest-lookup', +'connector' = 'http', 'format' = 'json', 'url' = 'http://localhost:8080/client', 'asyncPolling' = 'true' @@ -157,7 +159,7 @@ Note the options with the prefix _http_ are the HTTP connector specific options, | Option | Required | Description/Value | |:-----------------------------------------------------------------------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| connector | required | The Value should be set to _rest-lookup_ | +| connector | required | The Value should be set to _http_ | | format | required | Flink's format name that should be used to decode REST response, Use `json` for a typical REST endpoint. | | url | required | The base URL that should be use for GET requests. For example _http://localhost:8080/client_ | | asyncPolling | optional | true/false - determines whether Async Polling should be used. Mechanism is based on Flink's Async I/O. | @@ -218,13 +220,13 @@ The HTTP connector supplies a number of Query Creators that you can use define t -
generic-json-url
+
http-generic-json-url
✓ ✓ ✓ -
generic-get-query
+
http-generic-get-query
<✓ for GETs/td> ✓ for PUTs and POSTs @@ -233,20 +235,20 @@ The HTTP connector supplies a number of Query Creators that you can use define t -### generic-json-url Query Creator +### http-generic-json-url Query Creator -The recommended Query creator for json is called _generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports +The recommended Query creator for json is called _http-generic-json-url_, which allows column content to be mapped as URL, path, body and query parameter request values; it supports POST, PUT and GET operations. This query creator allows you to issue json requests without needing to code your own custom http connector. The mappings from columns to the json request are supplied in the query creator configuration parameters `http.request.query-param-fields`, `http.request.body-fields` and `http.request.url-map`. -### generic-json-url Query Creator +### http-generic-json-url Query Creator -The default Query Creator is called _generic-json-url_. For body based queries such as POST/PUT requests, the +The default Query Creator is called _http-generic-json-url_. For body based queries such as POST/PUT requests, the ([GenericGetQueryCreator](flink-connector-http/src/main/java/org/apache/flink/connectors/http/internal/table/lookup/querycreators/GenericGetQueryCreator.java))is provided as a default query creator. This implementation uses Flink's [json-format](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/) to convert RowData object into Json String. For GET requests can be used for query parameter based queries. -The _generic-json-url_ allows for using custom formats that will perform serialization to Json. Thanks to this, users can create their own logic for converting RowData to Json Strings suitable for their HTTP endpoints and use this logic as custom format +The _http-generic-json-url_ allows for using custom formats that will perform serialization to Json. Thanks to this, users can create their own logic for converting RowData to Json Strings suitable for their HTTP endpoints and use this logic as custom format with HTTP Lookup connector and SQL queries. To create a custom format user has to implement Flink's `SerializationSchema` and `SerializationFormatFactory` interfaces and register custom format factory along other factories in `resources/META-INF.services/org.apache.flink.table.factories.Factory` file. This is common Flink mechanism for providing custom implementations for various factories. @@ -266,7 +268,7 @@ CREATE TABLE http-lookup ( id bigint, some_field string ) WITH ( - 'connector' = 'rest-lookup', + 'connector' = 'http', 'format' = 'json', 'url' = 'http://localhost:8080/client', 'asyncPolling' = 'true', diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactory.java index 155053c6..f48ff0d9 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactory.java @@ -64,7 +64,7 @@ * CREATE TABLE httplookup ( * id bigint * ) with ( - * 'connector' = 'rest-lookup', + * 'connector' = 'http', * 'url' = 'http://example.com/myendpoint', * 'format' = 'json', * 'http.source.lookup.request-callback' = 'my-callback' diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/LookupQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/LookupQueryCreatorFactory.java index d3a6322b..04092f82 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/LookupQueryCreatorFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/LookupQueryCreatorFactory.java @@ -46,7 +46,7 @@ * id bigint, * some_field string * ) WITH ( - * 'connector' = 'rest-lookup', + * 'connector' = 'http', * 'format' = 'json', * 'url' = 'http://example.com/myendpoint', * 'http.source.lookup.query-creator' = 'my-query-creator' diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java index f9b07ab5..c5d0d3c3 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactory.java @@ -74,6 +74,8 @@ /** Http Lookup Table Source Factory. */ public class HttpLookupTableSourceFactory implements DynamicTableSourceFactory { + private static final String IDENTIFIER = "http"; + private static DataTypes.Field columnToField(Column column) { return FIELD( column.getName(), @@ -141,7 +143,7 @@ protected void validateHttpLookupSourceOptions(ReadableConfig tableOptions) @Override public String factoryIdentifier() { - return "rest-lookup"; + return IDENTIFIER; } @Override diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/Slf4jHttpLookupPostRequestCallbackFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/Slf4jHttpLookupPostRequestCallbackFactory.java index 39e839ed..5a8cae7a 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/Slf4jHttpLookupPostRequestCallbackFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/Slf4jHttpLookupPostRequestCallbackFactory.java @@ -29,7 +29,7 @@ public class Slf4jHttpLookupPostRequestCallbackFactory implements HttpPostRequestCallbackFactory { - public static final String IDENTIFIER = "slf4j-lookup-logger"; + public static final String IDENTIFIER = "http-slf4j-lookup-logger"; @Override public HttpPostRequestCallback createHttpPostRequestCallback() { diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/ElasticSearchLiteQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/ElasticSearchLiteQueryCreatorFactory.java index 6b0e9c2c..734525a3 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/ElasticSearchLiteQueryCreatorFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/ElasticSearchLiteQueryCreatorFactory.java @@ -29,7 +29,7 @@ /** Factory for creating {@link ElasticSearchLiteQueryCreator}. */ public class ElasticSearchLiteQueryCreatorFactory implements LookupQueryCreatorFactory { - public static final String IDENTIFIER = "elasticsearch-lite"; + public static final String IDENTIFIER = "http-elasticsearch-lite"; @Override public LookupQueryCreator createLookupQueryCreator( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericGetQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericGetQueryCreatorFactory.java index 2894ff12..63351de7 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericGetQueryCreatorFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericGetQueryCreatorFactory.java @@ -29,7 +29,7 @@ /** Factory for creating {@link GenericGetQueryCreator}. */ public class GenericGetQueryCreatorFactory implements LookupQueryCreatorFactory { - public static final String IDENTIFIER = "generic-get-query"; + public static final String IDENTIFIER = "http-generic-get-query"; @Override public LookupQueryCreator createLookupQueryCreator( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java index a718371f..c03ce7e1 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactory.java @@ -54,7 +54,7 @@ public class GenericJsonAndUrlQueryCreatorFactory implements LookupQueryCreatorFactory { private static final long serialVersionUID = 1L; - public static final String ID = "generic-json-url"; + public static final String ID = "http-generic-json-url"; public static final ConfigOption> REQUEST_QUERY_PARAM_FIELDS = key("http.request.query-param-fields") diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactory.java index 75cb22e2..bf7c97d0 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactory.java @@ -39,7 +39,7 @@ /** Factory for creating {@link GenericJsonQueryCreatorFactory}. */ public class GenericJsonQueryCreatorFactory implements LookupQueryCreatorFactory { - public static final String IDENTIFIER = "generic-json-query"; + public static final String IDENTIFIER = "http-generic-json-query"; @Override public LookupQueryCreator createLookupQueryCreator( diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java index 6d4b2d1e..d3ef903e 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactory.java @@ -38,7 +38,7 @@ /** Factory for creating {@link HttpDynamicSink}. */ public class HttpDynamicTableSinkFactory extends AsyncDynamicTableSinkFactory { - public static final String IDENTIFIER = "http-sink"; + public static final String IDENTIFIER = "http-async-sink"; @Override public DynamicTableSink createDynamicTableSink(Context context) { diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/Slf4jHttpPostRequestCallbackFactory.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/Slf4jHttpPostRequestCallbackFactory.java index 78d00bff..2dc9e97f 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/Slf4jHttpPostRequestCallbackFactory.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/sink/Slf4jHttpPostRequestCallbackFactory.java @@ -29,7 +29,7 @@ public class Slf4jHttpPostRequestCallbackFactory implements HttpPostRequestCallbackFactory { - public static final String IDENTIFIER = "slf4j-logger"; + public static final String IDENTIFIER = "http-slf4j-logger"; @Override public HttpPostRequestCallback createHttpPostRequestCallback() { diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactoryTest.java index ef20a3f5..f5bf9ac1 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactoryTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/HttpPostRequestCallbackFactoryTest.java @@ -147,7 +147,7 @@ public void httpLookupPostRequestCallbackFactoryTest() + " 'format' = 'json',\n" + " 'http.source.lookup.request-callback' = '%s'\n" + ")", - "rest-lookup", + "http", "http://localhost:" + SERVER_PORT + "/myendpoint", TEST_LOOKUP_POST_REQUEST_CALLBACK_IDENT); tEnv.executeSql(createTable2); diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java index 24153bc5..32be150a 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/StreamTableJob.java @@ -49,7 +49,7 @@ public static void main(String[] args) { + " 'fields.id2.kind' = 'sequence', 'fields.id2.start' = '2'," + " 'fields.id2.end' = '120')"); tableEnv.executeSql( - "CREATE TABLE Customers (id STRING, id2 STRING, msg STRING, uuid STRING, isActive STRING, balance STRING) WITH ('connector' = 'rest-lookup'," + "CREATE TABLE Customers (id STRING, id2 STRING, msg STRING, uuid STRING, isActive STRING, balance STRING) WITH ('connector' = 'http'," + " 'url' = 'http://localhost:" + WireMockServerPortAllocator.getServerPort() + "/client', " diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java index 08e031ab..6ac8dce1 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceFactoryTest.java @@ -97,7 +97,7 @@ void shouldCreateForMandatoryFields() { @Test void shouldThrowIfMissingUrl() { - Map options = Collections.singletonMap("connector", "rest-lookup"); + Map options = Collections.singletonMap("connector", "http"); assertThatExceptionOfType(ValidationException.class) .isThrownBy(() -> createTableSource(SCHEMA, options)); } @@ -120,7 +120,7 @@ void shouldHandleEmptyUrlArgs() { private Map getMandatoryOptions() { return Map.of( - "connector", "rest-lookup", + "connector", "http", "url", "http://localhost:" + WireMockServerPortAllocator.PORT_BASE + "/service", "format", "json"); } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java index ade60a03..1ffbf637 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceITCaseTest.java @@ -182,7 +182,7 @@ void testHttpLookupJoin(String methodName) throws Exception { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + ((StringUtils.isNullOrWhitespaceOnly(methodName)) ? "" : "'lookup-method' = '" + methodName + "',") @@ -222,7 +222,7 @@ void testHttpLookupJoinNoDataFromEndpoint() { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'url' = 'http://localhost:" + serverPort + "/client'," @@ -269,7 +269,7 @@ void testLookupWithRetry() throws Exception { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'url' = 'http://localhost:" + serverPort + "/client'," @@ -322,7 +322,7 @@ void testLookupIgnoreResponse() throws Exception { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'url' = 'http://localhost:" + serverPort + "/client'," @@ -362,7 +362,7 @@ void testHttpsMTlsLookupJoin() throws Exception { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'url' = 'https://localhost:" + secServerPort + "/client'," @@ -421,7 +421,7 @@ void testLookupJoinProjectionPushDown() throws Exception { + "'format' = 'json'," + "'lookup-request.format' = 'json'," + "'lookup-request.format.json.fail-on-missing-field' = 'true'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'POST'," + "'url' = 'http://localhost:" + serverPort @@ -497,7 +497,7 @@ void testLookupJoinProjectionPushDownNested() throws Exception { + "'format' = 'json'," + "'lookup-request.format' = 'json'," + "'lookup-request.format.json.fail-on-missing-field' = 'true'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'POST'," + "'url' = 'http://localhost:" + serverPort @@ -573,7 +573,7 @@ void testLookupJoinOnRowType() throws Exception { + "'format' = 'json'," + "'lookup-request.format' = 'json'," + "'lookup-request.format.json.fail-on-missing-field' = 'true'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'POST'," + "'url' = 'http://localhost:" + serverPort @@ -643,7 +643,7 @@ void testLookupJoinOnRowTypeAndRootColumn() throws Exception { + "'format' = 'json'," + "'lookup-request.format' = 'json'," + "'lookup-request.format.json.fail-on-missing-field' = 'true'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'POST'," + "'url' = 'http://localhost:" + serverPort @@ -753,7 +753,7 @@ private void testLookupJoinOnRowWithRowTypeImpl( + fields + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'POST'," + "'url' = 'http://localhost:" + serverPort @@ -863,7 +863,7 @@ void testNestedLookupJoinWithoutCast() throws Exception { + "'lookup-request.format' = 'json'," + "'lookup-request.format.json.fail-on-missing-field' = 'true'," + "'lookup-method' = 'POST'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'url' = 'http://localhost:" + serverPort + "/client'," @@ -933,7 +933,7 @@ void testHttpLookupJoinWithCache(boolean isAsync) throws Exception { + ">" + ") WITH (" + "'format' = 'json'," - + "'connector' = 'rest-lookup'," + + "'connector' = 'http'," + "'lookup-method' = 'GET'," + "'url' = 'http://localhost:" + serverPort @@ -1528,7 +1528,7 @@ private String createLookupTableSql(TestSpec spec) { .append("completionState STRING METADATA from 'http-completion-state'"); } - sql.append(") WITH (").append("'format' = 'json',").append("'connector' = 'rest-lookup',"); + sql.append(") WITH (").append("'format' = 'json',").append("'connector' = 'http',"); if (!StringUtils.isNullOrWhitespaceOnly(spec.methodName)) { sql.append("'lookup-method' = '").append(spec.methodName).append("',"); diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java index 7bae0f2c..429f30bd 100644 --- a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSourceTest.java @@ -345,7 +345,7 @@ private Map getOptionsWithAsync() { private Map getOptions() { return Map.of( - "connector", "rest-lookup", + "connector", "http", "url", "http://localhost:8080/service", "format", "json"); }