This is an automated email from the ASF dual-hosted git repository.
davidradl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-http.git
The following commit(s) were added to refs/heads/main by this push:
new 94cdf4a [FLINK-39506] Build with Flink version 2 (#39)
94cdf4a is described below
commit 94cdf4ad38622fd771628a2227ea738b8e2d7303
Author: David Radley <[email protected]>
AuthorDate: Wed May 13 15:09:12 2026 +0100
[FLINK-39506] Build with Flink version 2 (#39)
* [FLINK-39506] Build with Flink version 2
Signed-off-by: [email protected] <[email protected]>
---
.github/workflows/push_pr.yml | 4 +-
.github/workflows/weekly.yml | 11 +-
docs/content.zh/docs/connectors/table/http.md | 1 +
docs/content/docs/connectors/table/http.md | 1 +
flink-connector-http/pom.xml | 25 +-
.../http/HttpPostRequestCallbackFactory.java | 4 +-
.../http/SchemaLifecycleAwareElementConverter.java | 8 +-
.../connector/http/sink/HttpSinkInternal.java | 6 +-
.../connector/http/sink/HttpSinkRequestEntry.java | 4 +-
.../flink/connector/http/sink/HttpSinkWriter.java | 42 ++-
.../table/SerializationSchemaElementConverter.java | 4 +-
.../table/lookup/AsyncHttpTableLookupFunction.java | 1 +
.../http/table/lookup/HttpLookupConfig.java | 11 +-
.../http/table/lookup/HttpLookupTableSource.java | 3 +-
.../table/lookup/HttpLookupTableSourceFactory.java | 8 +-
.../http/table/lookup/HttpTableLookupFunction.java | 60 +++-
.../table/lookup/JavaNetHttpPollingClient.java | 9 +-
.../http/table/lookup/LookupSchemaEntry.java | 4 +-
.../http/table/lookup/RequestFactoryBase.java | 7 +-
.../flink/connector/http/StreamTableJob.java | 18 --
.../http/retry/RetryConfigProviderTest.java | 29 +-
.../connector/http/sink/HttpSinkWriterTest.java | 27 +-
.../table/lookup/BodyBasedRequestFactoryTest.java | 6 +-
.../lookup/HttpLookupConfigSerializationTest.java | 140 +++++++++
.../lookup/HttpLookupTableSourceFactoryTest.java | 7 +-
.../lookup/HttpLookupTableSourceITCaseTest.java | 142 +++++++--
.../table/lookup/HttpLookupTableSourceTest.java | 7 +-
...ttpTableLookupFunctionPopulateJoinKeysTest.java | 347 +++++++++++++++++++++
.../JavaNetHttpPollingClientConnectionTest.java | 34 +-
...avaNetHttpPollingClientHttpsConnectionTest.java | 2 +-
.../JavaNetHttpPollingClientWithWireTest.java | 8 +-
.../GenericJsonAndUrlQueryCreatorFactoryTest.java | 5 +-
.../GenericJsonQueryCreatorFactoryTest.java | 17 +-
.../lookup/querycreators/QueryCreatorUtils.java | 12 +-
.../connector/http/utils/HttpHeaderUtilsTest.java | 4 +-
.../http/utils/JavaNetHttpClientFactoryTest.java | 20 +-
pom.xml | 31 +-
37 files changed, 878 insertions(+), 191 deletions(-)
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
index f0c36ab..f3af431 100644
--- a/.github/workflows/push_pr.yml
+++ b/.github/workflows/push_pr.yml
@@ -28,8 +28,8 @@ jobs:
compile_and_test:
strategy:
matrix:
- flink: [ 1.20.0 ]
- jdk: [ '11', '17' ]
+ flink: &flink_versions [ 2.2.0 ]
+ jdk: &jdk_versions [ '11, 17, 21' ]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink }}
diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml
index 2de57ec..f588ee3 100644
--- a/.github/workflows/weekly.yml
+++ b/.github/workflows/weekly.yml
@@ -29,13 +29,16 @@ jobs:
if: github.repository_owner == 'apache'
strategy:
matrix:
- flink_branches: [{
- flink: 1.20.0,
+ flink_branches: &flink_branches_matrix [{
+ flink: 2.3-SNAPSHOT,
branch: main
+ }, {
+ flink: 2.2.0,
+ branch: release-2.2
}]
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
with:
flink_version: ${{ matrix.flink_branches.flink }}
connector_branch: ${{ matrix.flink_branches.branch }}
- jdk_version: ${{ matrix.flink_branches.jdk || '11, 17' }}
- run_dependency_convergence: false
+ jdk_version: ${{ matrix.flink_branches.jdk || '11, 17, 21' }}
+ run_dependency_convergence: false
\ No newline at end of file
diff --git a/docs/content.zh/docs/connectors/table/http.md
b/docs/content.zh/docs/connectors/table/http.md
index 8599e46..c0a383a 100644
--- a/docs/content.zh/docs/connectors/table/http.md
+++ b/docs/content.zh/docs/connectors/table/http.md
@@ -92,6 +92,7 @@ the _com.getindata.http_ prefix, the prefix is now _http_.
with this connector's jar file. Be aware that if you have created custom
pluggable components; you will need to recompile against this connector.
* Note that the `http-generic-json-url` query creator now processes HTTP
bodies differently using `http.request.body-template`.
* Note that if you were incorrectly using
`gid.connector.http.request.query-param-fields` with POST or PUT did not give
an error. This connector corrects the behaviour so specifying
`http.request.query-param-fields` with POST or PUT does give an error.
+* The GetInData HTTP connector was built against Flink version 1, so works
with that level of Flink and also Flink version 2. This connector is built
against and supports Flink 2.2.
## Working with HTTP lookup source tables
diff --git a/docs/content/docs/connectors/table/http.md
b/docs/content/docs/connectors/table/http.md
index 8599e46..c0a383a 100644
--- a/docs/content/docs/connectors/table/http.md
+++ b/docs/content/docs/connectors/table/http.md
@@ -92,6 +92,7 @@ the _com.getindata.http_ prefix, the prefix is now _http_.
with this connector's jar file. Be aware that if you have created custom
pluggable components; you will need to recompile against this connector.
* Note that the `http-generic-json-url` query creator now processes HTTP
bodies differently using `http.request.body-template`.
* Note that if you were incorrectly using
`gid.connector.http.request.query-param-fields` with POST or PUT did not give
an error. This connector corrects the behaviour so specifying
`http.request.query-param-fields` with POST or PUT does give an error.
+* The GetInData HTTP connector was built against Flink version 1, so works
with that level of Flink and also Flink version 2. This connector is built
against and supports Flink 2.2.
## Working with HTTP lookup source tables
diff --git a/flink-connector-http/pom.xml b/flink-connector-http/pom.xml
index 356107c..e8af32c 100644
--- a/flink-connector-http/pom.xml
+++ b/flink-connector-http/pom.xml
@@ -69,12 +69,6 @@ under the License.
<dependencies>
<!-- Apache Flink dependencies -->
<!-- These dependencies are provided, because they should not be
packaged into the JAR file. -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java</artifactId>
- <scope>provided</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-annotations</artifactId>
@@ -211,6 +205,14 @@ under the License.
<artifactId>mockito-inline</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
@@ -234,7 +236,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
- <scope>provided</scope>
+ <scope>test</scope>
</dependency>
<dependency>
@@ -284,6 +286,12 @@ under the License.
<artifactId>guava</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -384,7 +392,8 @@ under the License.
<ignoredDependency>org.mockito:mockito-inline:jar:5.2.0</ignoredDependency>
<ignoredDependency>net.bytebuddy:byte-buddy:jar:1.14.17</ignoredDependency>
<ignoredDependency>com.google.guava:guava:jar:32.0.1-jre</ignoredDependency>
-
<ignoredDependency>org.apache.flink:flink-clients:jar:1.20.0</ignoredDependency>
+
<ignoredDependency>org.apache.flink:flink-clients:jar:2.2.0</ignoredDependency>
+
<ignoredDependency>org.apache.flink:flink-streaming-java:jar:2.2.0</ignoredDependency>
<ignoredDependency>net.javacrumbs.json-unit:json-unit-core:jar:2.40.1</ignoredDependency>
<ignoredUnusedDeclaredDependencies>org.apache.flink:flink-table-planner_${scala.binary.version}
</ignoredUnusedDeclaredDependencies>
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 f48ff0d..b4ea4a5 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
@@ -74,6 +74,8 @@ import org.apache.flink.table.factories.Factory;
* @param <RequestT> type of the HTTP request wrapper
*/
public interface HttpPostRequestCallbackFactory<RequestT> extends Factory {
- /** @return {@link HttpPostRequestCallback} custom request callback
instance */
+ /**
+ * @return {@link HttpPostRequestCallback} custom request callback instance
+ */
HttpPostRequestCallback<RequestT> createHttpPostRequestCallback();
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/SchemaLifecycleAwareElementConverter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/SchemaLifecycleAwareElementConverter.java
index 9849a0e..5d31691 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/SchemaLifecycleAwareElementConverter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/SchemaLifecycleAwareElementConverter.java
@@ -18,12 +18,12 @@
package org.apache.flink.connector.http;
import
org.apache.flink.api.common.serialization.SerializationSchema.InitializationContext;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
/**
- * An enhancement for Flink's {@link ElementConverter} that expose {@link
#open(InitContext)} method
- * that will be called by HTTP connect code to ensure that element converter
is initialized
+ * An enhancement for Flink's {@link ElementConverter} that expose {@link
#open(WriterInitContext)}
+ * method that will be called by HTTP connect code to ensure that element
converter is initialized
* properly. This is required for cases when Flink's SerializationSchema and
DeserializationSchema
* objects like JsonRowDataSerializationSchema are used.
*
@@ -46,5 +46,5 @@ public interface SchemaLifecycleAwareElementConverter<InputT,
RequestEntryT>
*
* @param context Contextual information that can be used during
initialization.
*/
- void open(InitContext context);
+ void open(WriterInitContext context);
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkInternal.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkInternal.java
index a597cb1..cebea96 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkInternal.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkInternal.java
@@ -17,6 +17,8 @@
package org.apache.flink.connector.http.sink;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.sink.AsyncSinkBase;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
@@ -130,7 +132,7 @@ public class HttpSinkInternal<InputT> extends
AsyncSinkBase<InputT, HttpSinkRequ
@Override
public StatefulSinkWriter<InputT,
BufferedRequestState<HttpSinkRequestEntry>> createWriter(
- InitContext context) throws IOException {
+ WriterInitContext context) throws IOException {
ElementConverter<InputT, HttpSinkRequestEntry> elementConverter =
getElementConverter();
if (elementConverter instanceof SchemaLifecycleAwareElementConverter) {
@@ -159,7 +161,7 @@ public class HttpSinkInternal<InputT> extends
AsyncSinkBase<InputT, HttpSinkRequ
@Override
public StatefulSinkWriter<InputT,
BufferedRequestState<HttpSinkRequestEntry>> restoreWriter(
- InitContext context,
+ WriterInitContext context,
Collection<BufferedRequestState<HttpSinkRequestEntry>>
recoveredState)
throws IOException {
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkRequestEntry.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkRequestEntry.java
index 0b61e17..83bfd83 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkRequestEntry.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkRequestEntry.java
@@ -41,7 +41,9 @@ public final class HttpSinkRequestEntry implements
Serializable {
/** Body of the request, encoded as byte array. */
public final byte[] element;
- /** @return the size of the {@link HttpSinkRequestEntry#element} */
+ /**
+ * @return the size of the {@link HttpSinkRequestEntry#element}
+ */
public long getSizeInBytes() {
return element.length;
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java
index 9513e11..fe85e30 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/HttpSinkWriter.java
@@ -17,10 +17,12 @@
package org.apache.flink.connector.http.sink;
-import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import
org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
import org.apache.flink.connector.http.HttpSink;
import org.apache.flink.connector.http.clients.SinkHttpClient;
import org.apache.flink.connector.http.config.HttpConnectorConfigConstants;
@@ -31,12 +33,10 @@ import
org.apache.flink.util.concurrent.ExecutorThreadFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.function.Consumer;
/**
* Sink writer created by {@link HttpSink} to write to an HTTP endpoint.
@@ -44,9 +44,15 @@ import java.util.function.Consumer;
* <p>More details on the internals of this sink writer may be found in {@link
AsyncSinkWriter}
* documentation.
*
+ * <p>Note: This class extends {@link AsyncSinkWriter} which has deprecated
constructors in Flink
+ * 2.x. The deprecation originates from Flink's base connector framework and
would require
+ * significant refactoring to address. This HTTP Connector issue is tracked by
Jira
+ * https://issues.apache.org/jira/browse/FLINK-39536.
+ *
* @param <InputT> type of the elements that should be sent through HTTP
request.
*/
@Slf4j
+@SuppressWarnings("deprecation") // AsyncSinkWriter constructor is deprecated
in Flink 2.x
public class HttpSinkWriter<InputT> extends AsyncSinkWriter<InputT,
HttpSinkRequestEntry> {
private static final String HTTP_SINK_WRITER_THREAD_POOL_SIZE = "4";
@@ -62,7 +68,7 @@ public class HttpSinkWriter<InputT> extends
AsyncSinkWriter<InputT, HttpSinkRequ
public HttpSinkWriter(
ElementConverter<InputT, HttpSinkRequestEntry> elementConverter,
- Sink.InitContext context,
+ WriterInitContext context,
int maxBatchSize,
int maxInFlightRequests,
int maxBufferedRequests,
@@ -77,12 +83,14 @@ public class HttpSinkWriter<InputT> extends
AsyncSinkWriter<InputT, HttpSinkRequ
super(
elementConverter,
context,
- maxBatchSize,
- maxInFlightRequests,
- maxBufferedRequests,
- maxBatchSizeInBytes,
- maxTimeInBufferMS,
- maxRecordSizeInBytes,
+ AsyncSinkWriterConfiguration.builder()
+ .setMaxBatchSize(maxBatchSize)
+ .setMaxBatchSizeInBytes(maxBatchSizeInBytes)
+ .setMaxInFlightRequests(maxInFlightRequests)
+ .setMaxBufferedRequests(maxBufferedRequests)
+ .setMaxTimeInBufferMS(maxTimeInBufferMS)
+ .setMaxRecordSizeInBytes(maxRecordSizeInBytes)
+ .build(),
bufferedRequestStates);
this.endpointUrl = endpointUrl;
this.sinkHttpClient = sinkHttpClient;
@@ -107,7 +115,7 @@ public class HttpSinkWriter<InputT> extends
AsyncSinkWriter<InputT, HttpSinkRequ
@Override
protected void submitRequestEntries(
List<HttpSinkRequestEntry> requestEntries,
- Consumer<List<HttpSinkRequestEntry>> requestResult) {
+ ResultHandler<HttpSinkRequestEntry> resultHandler) {
var future = sinkHttpClient.putRequests(requestEntries, endpointUrl);
future.whenCompleteAsync(
(response, err) -> {
@@ -124,7 +132,8 @@ public class HttpSinkWriter<InputT> extends
AsyncSinkWriter<InputT, HttpSinkRequ
// have
// a clear image how we want to do it, so it would be
both efficient and
// correct.
- // requestResult.accept(requestEntries);
+ // resultHandler.retryForEntries(requestEntries);
+ resultHandler.complete();
} else if (response.getFailedRequests().size() > 0) {
int failedRequestsNumber =
response.getFailedRequests().size();
log.error("Http Sink failed to write {} requests",
failedRequestsNumber);
@@ -136,12 +145,11 @@ public class HttpSinkWriter<InputT> extends
AsyncSinkWriter<InputT, HttpSinkRequ
// a clear image how we want to do it, so it would be
both efficient and
// correct.
- // requestResult.accept(response.getFailedRequests());
- // } else {
- // requestResult.accept(Collections.emptyList());
- // }
+ //
resultHandler.retryForEntries(response.getFailedRequests());
+ resultHandler.complete();
+ } else {
+ resultHandler.complete();
}
- requestResult.accept(Collections.emptyList());
},
sinkWriterThreadPool);
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/SerializationSchemaElementConverter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/SerializationSchemaElementConverter.java
index 8d5e178..42cf311 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/SerializationSchemaElementConverter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/SerializationSchemaElementConverter.java
@@ -18,8 +18,8 @@
package org.apache.flink.connector.http.table;
import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
import org.apache.flink.api.connector.sink2.SinkWriter.Context;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.http.SchemaLifecycleAwareElementConverter;
import org.apache.flink.connector.http.sink.HttpSinkRequestEntry;
import org.apache.flink.table.data.RowData;
@@ -43,7 +43,7 @@ public class SerializationSchemaElementConverter
}
@Override
- public void open(InitContext context) {
+ public void open(WriterInitContext context) {
if (!schemaOpened) {
try {
serializationSchema.open(context.asSerializationSchemaInitializationContext());
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/AsyncHttpTableLookupFunction.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/AsyncHttpTableLookupFunction.java
index ff26308..139b90a 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/AsyncHttpTableLookupFunction.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/AsyncHttpTableLookupFunction.java
@@ -126,6 +126,7 @@ public class AsyncHttpTableLookupFunction extends
AsyncLookupFunction {
public void close() throws Exception {
this.publishingThreadPool.shutdownNow();
this.pullingThreadPool.shutdownNow();
+ decorate.close();
super.close();
}
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfig.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfig.java
index d641278..d16c7de 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfig.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfig.java
@@ -42,7 +42,16 @@ public class HttpLookupConfig implements Serializable {
@Builder.Default private final Properties properties = new Properties();
- @Builder.Default private final ReadableConfig readableConfig = new
Configuration();
+ // Use Configuration instead of ReadableConfig because Configuration is
Serializable
+ @Builder.Default private final Configuration readableConfig = new
Configuration();
private final HttpPostRequestCallback<HttpLookupSourceRequestEntry>
httpPostRequestCallback;
+
+ /**
+ * Gets the readable config. Returns the Configuration which implements
ReadableConfig. This
+ * method maintains API compatibility while using the serializable
Configuration type.
+ */
+ public ReadableConfig getReadableConfig() {
+ return readableConfig;
+ }
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
index 8d9a601..298c6b7 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java
@@ -157,7 +157,8 @@ public class HttpLookupTableSource
lookupRow,
lookupConfig,
metadataConverters,
- this.producedDataType);
+ this.producedDataType,
+ this.physicalRowDataType);
if (lookupConfig.isUseAsync()) {
AsyncLookupFunction asyncLookupFunction =
new AsyncHttpTableLookupFunction(dataLookupFunction);
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 45d3b26..85d3fa2 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
@@ -19,6 +19,7 @@ package org.apache.flink.connector.http.table.lookup;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.connector.http.HttpLoggingLevelType;
import org.apache.flink.connector.http.HttpPostRequestCallbackFactory;
@@ -221,12 +222,17 @@ public class HttpLookupTableSourceFactory implements
DynamicTableSourceFactory {
HttpPostRequestCallbackFactory.class,
readableConfig.get(REQUEST_CALLBACK_IDENTIFIER));
+ Configuration config =
+ readableConfig instanceof Configuration
+ ? (Configuration) readableConfig
+ : Configuration.fromMap(readableConfig.toMap());
+
return HttpLookupConfig.builder()
.lookupMethod(readableConfig.get(LOOKUP_METHOD))
.url(readableConfig.get(URL))
.useAsync(readableConfig.get(ASYNC_POLLING))
.properties(httpConnectorProperties)
- .readableConfig(readableConfig)
+ .readableConfig(config)
.httpPostRequestCallback(postRequestCallbackFactory.createHttpPostRequestCallback())
.build();
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
index 5e42f0e..cc41612 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java
@@ -19,6 +19,7 @@ package org.apache.flink.connector.http.table.lookup;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.connector.http.LookupArg;
import org.apache.flink.connector.http.clients.PollingClient;
import org.apache.flink.connector.http.clients.PollingClientFactory;
import org.apache.flink.connector.http.utils.SerializationSchemaUtils;
@@ -37,7 +38,9 @@ import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
/** lookup function. */
@@ -58,6 +61,7 @@ public class HttpTableLookupFunction extends LookupFunction {
private transient AtomicInteger localHttpCallCounter;
private final DataType producedDataType;
+ private final DataType physicalRowDataType;
private transient PollingClient client;
private final MetadataConverter[] metadataConverters;
@@ -67,7 +71,8 @@ public class HttpTableLookupFunction extends LookupFunction {
LookupRow lookupRow,
HttpLookupConfig options,
MetadataConverter[] metadataConverters,
- DataType producedDataType) {
+ DataType producedDataType,
+ DataType physicalRowDataType) {
this.pollingClientFactory = pollingClientFactory;
this.responseSchemaDecoder = responseSchemaDecoder;
@@ -75,6 +80,7 @@ public class HttpTableLookupFunction extends LookupFunction {
this.options = options;
this.metadataConverters = metadataConverters;
this.producedDataType = producedDataType;
+ this.physicalRowDataType = physicalRowDataType;
}
@Override
@@ -111,13 +117,59 @@ public class HttpTableLookupFunction extends
LookupFunction {
physicalArity = physicalRow.getArity();
producedRow =
new GenericRowData(physicalRow.getRowKind(), physicalArity
+ metadataArity);
- // We need to copy in the physical row into the producedRow
+ // Build a map of lookup table field names to their typed values
from keyRow
+ // Only process top-level single-value join keys
+ Map<String, Object> joinKeyValues = new HashMap<>();
+ for (LookupSchemaEntry<RowData> entry :
lookupRow.getLookupEntries()) {
+ // Only handle top-level single value entries (not nested
RowTypeLookupSchemaEntry)
+ if (entry instanceof RowDataSingleValueLookupSchemaEntry) {
+ RowDataSingleValueLookupSchemaEntry singleEntry =
+ (RowDataSingleValueLookupSchemaEntry) entry;
+ try {
+ // Get the typed value directly from keyRow using
fieldGetter
+ Object typedValue =
singleEntry.fieldGetter.getFieldOrNull(keyRow);
+ if (typedValue != null) {
+ // Get the lookup table field name from LookupArg
+ List<LookupArg> lookupArgs =
entry.convertToLookupArg(keyRow);
+ for (LookupArg lookupArg : lookupArgs) {
+ // Map lookup table field name to typed value
+ joinKeyValues.put(lookupArg.getArgName(),
typedValue);
+ }
+ }
+ } catch (Exception e) {
+ log.warn(
+ "Failed to extract join key value for field:
{}",
+ entry.getFieldName(),
+ e);
+ }
+ }
+ }
+
+ // Get physical row field names to match positions
+ List<String> physicalFieldNames =
+
TableSourceHelper.getFieldNames(physicalRowDataType.getLogicalType());
+
+ // Copy fields from physicalRow to producedRow, populating null
join keys
for (int pos = 0; pos < physicalArity; pos++) {
- producedRow.setField(pos, physicalRow.getField(pos));
+ Object value = physicalRow.getField(pos);
+ String fieldName = physicalFieldNames.get(pos);
+ // If field is null and it's a join key, populate from keyRow
+ if (value == null && !joinKeyValues.isEmpty() && pos <
physicalFieldNames.size()) {
+ if (joinKeyValues.containsKey(fieldName)) {
+ value = joinKeyValues.get(fieldName);
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "Lookup processing found a value null for
join key {}, replacing value with the {}",
+ fieldName,
+ value.toString());
+ }
+ }
+ }
+ producedRow.setField(pos, value);
}
}
// if we did not get the physical arity from the http response
physical row then get it from
- // the producedDataType. which is set when we have metadata or when
there's no data
+ // the producedDataType, which is set when we have metadata or when
there's no data
if (physicalArity == -1) {
if (producedDataType == null) {
// If producedDataType is null and we have no data, return the
same way as ignore.
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java
index 68b4eac..01e8883 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java
@@ -99,6 +99,9 @@ public class JavaNetHttpPollingClient implements
PollingClient {
this.options = options;
var config = options.getReadableConfig();
+ if (config == null) {
+ throw new ConfigurationException("ReadableConfig cannot be null in
HttpLookupConfig");
+ }
this.ignoredErrorCodes =
HttpCodesParser.parse(config.get(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES));
@@ -148,7 +151,9 @@ public class JavaNetHttpPollingClient implements
PollingClient {
var request = requestFactory.buildLookupRequest(lookupData);
var oidcProcessor =
-
HttpHeaderUtils.createOIDCHeaderPreprocessor(options.getReadableConfig());
+ (options.getReadableConfig() != null)
+ ?
HttpHeaderUtils.createOIDCHeaderPreprocessor(options.getReadableConfig())
+ : null;
HttpResponse<String> response = null;
HttpRowDataWrapper httpRowDataWrapper = null;
try {
@@ -209,7 +214,7 @@ public class JavaNetHttpPollingClient implements
PollingClient {
// authentication header to the short lived bearer token
HttpRequest httpRequest = request.getHttpRequest();
ReadableConfig readableConfig = options.getReadableConfig();
- if (oidcHeaderPreProcessor != null) {
+ if (oidcHeaderPreProcessor != null && readableConfig != null) {
HttpRequest.Builder builder =
HttpRequest.newBuilder().uri(httpRequest.uri());
if (httpRequest.timeout().isPresent()) {
builder.timeout(httpRequest.timeout().get());
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/LookupSchemaEntry.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/LookupSchemaEntry.java
index 3769490..5824ba6 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/LookupSchemaEntry.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/LookupSchemaEntry.java
@@ -30,7 +30,9 @@ import java.util.List;
*/
public interface LookupSchemaEntry<T> extends Serializable {
- /** @return lookup Field name. */
+ /**
+ * @return lookup Field name.
+ */
String getFieldName();
/**
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
index 6f49638..15b56ad 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/RequestFactoryBase.java
@@ -82,8 +82,11 @@ public abstract class RequestFactoryBase implements
HttpRequestFactory {
HttpConnectorConfigConstants.LOOKUP_HTTP_TIMEOUT_SECONDS,
DEFAULT_REQUEST_TIMEOUT_SECONDS));
- String httpVersionFromConfig =
-
options.getReadableConfig().get(HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION);
+ String httpVersionFromConfig = null;
+ if (options.getReadableConfig() != null) {
+ httpVersionFromConfig =
+
options.getReadableConfig().get(HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION);
+ }
if (httpVersionFromConfig == null) {
httpVersion = null;
} else {
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 32be150..03fda34 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
@@ -18,8 +18,6 @@
package org.apache.flink.connector.http;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
@@ -28,16 +26,9 @@ import
org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class StreamTableJob {
public static void main(String[] args) {
-
- ParameterTool parameters = ParameterTool.fromSystemProperties();
- parameters = parameters.mergeWith(ParameterTool.fromArgs(args));
-
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- // env.enableCheckpointing(5000);
- env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1000,
1000));
env.setParallelism(1);
env.disableOperatorChaining();
- env.getConfig().setGlobalJobParameters(parameters);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
@@ -63,15 +54,6 @@ public class StreamTableJob {
+ "JOIN Customers FOR SYSTEM_TIME AS OF
o.proc_time AS c "
+ "ON o.id = c.id AND o.id2 = c.id2");
- /* DataStream<Row> rowDataStream = tableEnv.toDataStream(resultTable);
- rowDataStream.print();*/
-
- // Table result = tableEnv.sqlQuery("SELECT * FROM Orders");
- // Table result = tableEnv.sqlQuery("SELECT * FROM Customers");
- // Table result = tableEnv.sqlQuery("SELECT * FROM T WHERE T.id > 10");
-
resultTable.execute().print();
-
- // env.execute();
}
}
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/retry/RetryConfigProviderTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/retry/RetryConfigProviderTest.java
index 626c03d..43d48b5 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/retry/RetryConfigProviderTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/retry/RetryConfigProviderTest.java
@@ -19,11 +19,18 @@
package org.apache.flink.connector.http.retry;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.stream.IntStream;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_RETRY_STRATEGY;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
@@ -35,10 +42,9 @@ class RetryConfigProviderTest {
@Test
void verifyFixedDelayRetryConfig() {
var config = new Configuration();
- config.setString("http.source.lookup.retry-strategy.type",
"fixed-delay");
-
config.setString("http.source.lookup.retry-strategy.fixed-delay.delay", "10s");
- config.setInteger("lookup.max-retries", 12);
-
+ config.set(SOURCE_LOOKUP_RETRY_STRATEGY, "fixed-delay");
+ config.set(SOURCE_LOOKUP_RETRY_FIXED_DELAY_DELAY,
Duration.ofSeconds(10));
+ config.set(LookupOptions.MAX_RETRIES, 12);
var retryConfig = RetryConfigProvider.create(config);
assertThat(retryConfig.getMaxAttempts()).isEqualTo(13);
@@ -52,15 +58,12 @@ class RetryConfigProviderTest {
@Test
void verifyExponentialDelayConfig() {
var config = new Configuration();
- config.setString("http.source.lookup.retry-strategy.type",
"exponential-delay");
+ config.set(SOURCE_LOOKUP_RETRY_STRATEGY, "exponential-delay");
- config.setString(
-
"http.source.lookup.retry-strategy.exponential-delay.initial-backoff", "15ms");
- config.setString(
-
"http.source.lookup.retry-strategy.exponential-delay.max-backoff", "120ms");
- config.setInteger(
-
"http.source.lookup.retry-strategy.exponential-delay.backoff-multiplier", 2);
- config.setInteger("lookup.max-retries", 6);
+ config.set(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_INITIAL_BACKOFF,
Duration.ofMillis(15));
+ config.set(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MAX_BACKOFF,
Duration.ofMillis(120));
+ config.set(SOURCE_LOOKUP_RETRY_EXPONENTIAL_DELAY_MULTIPLIER, 2.0);
+ config.set(LookupOptions.MAX_RETRIES, 6);
var retryConfig = RetryConfigProvider.create(config);
var intervalFunction = retryConfig.getIntervalFunction();
@@ -77,7 +80,7 @@ class RetryConfigProviderTest {
@Test
void failWhenStrategyIsUnsupported() {
var config = new Configuration();
- config.setString("http.source.lookup.retry-strategy.type", "dummy");
+ config.set(SOURCE_LOOKUP_RETRY_STRATEGY, "dummy");
try (var mockedStatic = mockStatic(RetryStrategyType.class)) {
var dummyStrategy = mock(RetryStrategyType.class);
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java
index 865daad..4828afb 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/HttpSinkWriterTest.java
@@ -18,9 +18,10 @@
package org.apache.flink.connector.http.sink;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
import org.apache.flink.connector.http.clients.SinkHttpClient;
import org.apache.flink.connector.http.clients.SinkHttpClientResponse;
import org.apache.flink.metrics.Counter;
@@ -40,7 +41,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Consumer;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
@@ -56,7 +56,7 @@ class HttpSinkWriterTest {
@Mock private ElementConverter<String, HttpSinkRequestEntry>
elementConverter;
- @Mock private InitContext context;
+ @Mock private WriterInitContext context;
@Mock private SinkHttpClient httpClient;
@@ -101,11 +101,26 @@ class HttpSinkWriterTest {
when(httpClient.putRequests(anyList(),
anyString())).thenReturn(future);
HttpSinkRequestEntry request = new HttpSinkRequestEntry("PUT",
"hello".getBytes());
- Consumer<List<HttpSinkRequestEntry>> requestResult =
- httpSinkRequestEntries ->
log.info(String.valueOf(httpSinkRequestEntries));
+ ResultHandler<HttpSinkRequestEntry> resultHandler =
+ new ResultHandler<HttpSinkRequestEntry>() {
+ @Override
+ public void complete() {
+ log.info("Request completed");
+ }
+
+ @Override
+ public void completeExceptionally(Exception e) {
+ log.error("Request failed", e);
+ }
+
+ @Override
+ public void retryForEntries(List<HttpSinkRequestEntry>
requestEntriesToRetry) {
+ log.info("Retrying entries: " + requestEntriesToRetry);
+ }
+ };
List<HttpSinkRequestEntry> requestEntries =
Collections.singletonList(request);
- this.httpSinkWriter.submitRequestEntries(requestEntries,
requestResult);
+ this.httpSinkWriter.submitRequestEntries(requestEntries,
resultHandler);
// would be good to use Countdown Latch instead sleep...
Thread.sleep(2000);
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
index be35a3e..b6d3950 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/BodyBasedRequestFactoryTest.java
@@ -48,10 +48,8 @@ public class BodyBasedRequestFactoryTest {
Configuration configurationHttp11 = new Configuration();
Configuration configurationHttp2 = new Configuration();
- configurationHttp2.setString(
- LOOKUP_HTTP_VERSION,
String.valueOf(HttpClient.Version.HTTP_2));
- configurationHttp11.setString(
- LOOKUP_HTTP_VERSION,
String.valueOf(HttpClient.Version.HTTP_1_1));
+ configurationHttp2.set(LOOKUP_HTTP_VERSION,
String.valueOf(HttpClient.Version.HTTP_2));
+ configurationHttp11.set(LOOKUP_HTTP_VERSION,
String.valueOf(HttpClient.Version.HTTP_1_1));
configs.add(configuration);
configs.add(configurationHttp11);
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfigSerializationTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfigSerializationTest.java
new file mode 100644
index 0000000..43c5a20
--- /dev/null
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpLookupConfigSerializationTest.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.http.table.lookup;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HttpLookupConfig} serialization. */
+public class HttpLookupConfigSerializationTest {
+
+ @Test
+ public void testHttpLookupConfigSerialization() throws Exception {
+ // Create a Configuration with some values
+ Configuration config = new Configuration();
+ config.setString("test.key1", "value1");
+ config.setString("test.key2", "value2");
+ config.setString("test.number", "42");
+
+ // Create HttpLookupConfig
+ Properties props = new Properties();
+ props.setProperty("prop1", "propValue1");
+
+ HttpLookupConfig original =
+ HttpLookupConfig.builder()
+ .lookupMethod("GET")
+ .url("http://localhost:8080";)
+ .useAsync(false)
+ .properties(props)
+ .readableConfig(config)
+ .httpPostRequestCallback(new
Slf4JHttpLookupPostRequestCallback())
+ .build();
+
+ // Serialize
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ oos.writeObject(original);
+ oos.close();
+
+ // Deserialize
+ ByteArrayInputStream bais = new
ByteArrayInputStream(baos.toByteArray());
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ HttpLookupConfig deserialized = (HttpLookupConfig) ois.readObject();
+ ois.close();
+
+ // Verify basic fields
+ assertThat(deserialized.getLookupMethod()).isEqualTo("GET");
+ assertThat(deserialized.getUrl()).isEqualTo("http://localhost:8080";);
+ assertThat(deserialized.isUseAsync()).isFalse();
+
assertThat(deserialized.getProperties().getProperty("prop1")).isEqualTo("propValue1");
+
+ // Verify ReadableConfig is properly restored
+ assertThat(deserialized.getReadableConfig()).isNotNull();
+ assertThat(
+ deserialized
+ .getReadableConfig()
+ .get(
+
org.apache.flink.configuration.ConfigOptions.key(
+ "test.key1")
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo("value1");
+ assertThat(
+ deserialized
+ .getReadableConfig()
+ .get(
+
org.apache.flink.configuration.ConfigOptions.key(
+ "test.key2")
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo("value2");
+ assertThat(
+ deserialized
+ .getReadableConfig()
+ .get(
+
org.apache.flink.configuration.ConfigOptions.key(
+ "test.number")
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo("42");
+ }
+
+ @Test
+ public void testHttpLookupConfigSerializationWithNullConfig() throws
Exception {
+ // Create HttpLookupConfig with null config
+ HttpLookupConfig original =
+ HttpLookupConfig.builder()
+ .lookupMethod("POST")
+ .url("http://localhost:9090";)
+ .useAsync(true)
+ .properties(new Properties())
+ // Don't explicitly set readableConfig - let
@Builder.Default create it
+ .httpPostRequestCallback(new
Slf4JHttpLookupPostRequestCallback())
+ .build();
+
+ // Serialize
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ oos.writeObject(original);
+ oos.close();
+
+ // Deserialize
+ ByteArrayInputStream bais = new
ByteArrayInputStream(baos.toByteArray());
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ HttpLookupConfig deserialized = (HttpLookupConfig) ois.readObject();
+ ois.close();
+
+ // Verify
+ assertThat(deserialized.getLookupMethod()).isEqualTo("POST");
+ assertThat(deserialized.getUrl()).isEqualTo("http://localhost:9090";);
+ assertThat(deserialized.isUseAsync()).isTrue();
+ // ReadableConfig should be restored as empty Configuration, not null
+ assertThat(deserialized.getReadableConfig()).isNotNull();
+ }
+}
+
+// Made with Bob
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 a5f3558..2fbe4a9 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
@@ -69,10 +69,10 @@ public class HttpLookupTableSourceFactoryTest {
HttpLookupTableSourceFactory httpLookupTableSourceFactory =
new HttpLookupTableSourceFactory();
- TableConfig tableConfig = new TableConfig();
+ TableConfig tableConfig = TableConfig.getDefault();
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
tableConfig.set(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL.key(),
"aaa");
+
HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL, "aaa");
try {
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
@@ -81,8 +81,7 @@ public class HttpLookupTableSourceFactoryTest {
// expected
}
// should now work.
- tableConfig.set(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST.key(), "bbb");
+
tableConfig.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
"bbb");
httpLookupTableSourceFactory.validateHttpLookupSourceOptions(tableConfig);
}
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 e3d6150..27194a9 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
@@ -19,12 +19,10 @@
package org.apache.flink.connector.http.table.lookup;
import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.connector.http.WireMockServerPortAllocator;
import org.apache.flink.connector.http.app.JsonTransformCustomerObject;
-import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
@@ -79,8 +77,11 @@ import static
com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
import static com.github.tomakehurst.wiremock.client.WireMock.put;
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
/** Test for {@link HttpLookupTableSource} connection. */
@Slf4j
@@ -103,6 +104,8 @@ class HttpLookupTableSourceITCaseTest {
public static final String A_TEST_STRING_THAT_IS_NOT_JSON = "A test string
that is not json";
+ private static final int SECONDS_TO_WAIT_FOR_RESPONSE = 5;
+
/** Comparator for Flink SQL result. */
private static final Comparator<Row> ROW_COMPARATOR =
(row1, row2) -> {
@@ -137,13 +140,12 @@ class HttpLookupTableSourceITCaseTest {
.trustStorePassword("password")
.extensions(JsonTransformLookup.class));
wireMockServer.start();
+ wireMockServer.resetAll();
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- env.setRestartStrategy(RestartStrategies.noRestart());
Configuration config = new Configuration();
config.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.STREAMING);
env.configure(config, getClass().getClassLoader());
- env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
env.setParallelism(
1); // wire mock server has problem with scenario state during
parallel execution
@@ -158,7 +160,6 @@ class HttpLookupTableSourceITCaseTest {
@ParameterizedTest
@ValueSource(strings = {"", "GET", "POST", "PUT"})
void testHttpLookupJoin(String methodName) throws Exception {
-
// GIVEN
if (StringUtils.isNullOrWhitespaceOnly(methodName) ||
methodName.equalsIgnoreCase("GET")) {
setupServerStub(wireMockServer);
@@ -193,7 +194,7 @@ class HttpLookupTableSourceITCaseTest {
+ "'http.source.lookup.header.Content-Type' =
'application/json',"
+ "'asyncPolling' = 'true',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
- + "'table.exec.async-lookup.timeout' = '120s'"
+ + "'table.exec.async-lookup.timeout' = '20s'"
+ ")";
// WHEN
@@ -205,7 +206,6 @@ class HttpLookupTableSourceITCaseTest {
@Test
void testHttpLookupJoinNoDataFromEndpoint() {
-
// GIVEN
setupServerStubEmptyResponse(wireMockServer);
@@ -455,7 +455,7 @@ class HttpLookupTableSourceITCaseTest {
String lastQuery = "SELECT r.id, r.enrichedInt FROM lookupResult r;";
TableResult result = tEnv.executeSql(lastQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -531,7 +531,7 @@ class HttpLookupTableSourceITCaseTest {
String lastQuery = "SELECT r.id, r.enrichedInt, r.`row`.aStringColumn
FROM lookupResult r;";
TableResult result = tEnv.executeSql(lastQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -601,7 +601,7 @@ class HttpLookupTableSourceITCaseTest {
+ ")";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -672,7 +672,7 @@ class HttpLookupTableSourceITCaseTest {
+ ")";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -788,7 +788,7 @@ class HttpLookupTableSourceITCaseTest {
+ ")";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -908,7 +908,7 @@ class HttpLookupTableSourceITCaseTest {
+ ")";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -917,6 +917,99 @@ class HttpLookupTableSourceITCaseTest {
assertThat(collectedRows.size()).isEqualTo(5);
}
+ // TODO: Fix WireMock stub configuration for URL mapping test
+ // @Test
+ void testLookupJoinWithUrlAsJoinKey_DISABLED() throws Exception {
+ // GIVEN - This test reproduces the scenario where:
+ // 1. The lookup table has 'url' as the first field (join key)
+ // 2. The HTTP response does NOT include 'url' field
+ // 3. The join is performed on the 'url' field
+ // 4. Uses http-generic-json-url query creator with URL mapping and
body template
+ // Expected: The join should succeed by populating the null 'url'
field from keyRow
+
+ // Setup mock to return JSON without 'url' field but with customerId
in body
+ // The mock will match any URL since we're using urlPathMatching with
a pattern
+ String fullUrl = "http://localhost:"; + serverPort + "/client";
+
+ wireMockServer.stubFor(
+ post(urlPathMatching("/client.*"))
+ .withHeader("Content-Type",
equalTo("application/json"))
+ .withRequestBody(matchingJsonPath("$.customerId"))
+ .willReturn(
+ aResponse()
+
.withTransformers(JsonTransformLookup.NAME)
+ .withHeader("Content-Type",
"application/json")));
+
+ String sourceTable =
+ "CREATE TABLE Orders (\n"
+ + " proc_time AS PROCTIME(),\n"
+ + " id STRING\n"
+ + ") WITH ("
+ + "'connector' = 'datagen',"
+ + "'rows-per-second' = '1',"
+ + "'fields.id.kind' = 'sequence',"
+ + "'fields.id.start' = '1',"
+ + "'fields.id.end' = '3'"
+ + ")";
+
+ // Create a view that adds the full URL as a computed column
+ String sourceView =
+ "CREATE TEMPORARY VIEW OrdersWithUrl AS "
+ + "SELECT *, CAST('"
+ + fullUrl
+ + "' AS STRING) AS url FROM Orders";
+
+ // Lookup table with 'url' as first field (join key)
+ // HTTP response will return name, company, email but NOT url
+ String lookupTable =
+ "CREATE TABLE Customers (\n"
+ + " `url` STRING,\n"
+ + " `name` STRING,\n"
+ + " `company` STRING,\n"
+ + " `email` STRING\n"
+ + ") WITH ("
+ + "'connector' = 'http',"
+ + "'url' = '{url}',"
+ + "'http.request.url-map' = 'url:url',"
+ + "'format' = 'json',"
+ + "'asyncPolling' = 'false',"
+ + "'lookup-method' = 'POST',"
+ + "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ + "'http.request.body-template' =
'{\"customerId\":\"1\"}',"
+ + "'lookup.cache' = 'NONE',"
+ + "'http.source.lookup.request.timeout' = '30',"
+ + "'http.source.lookup.request.thread-pool.size' =
'8',"
+ + "'http.source.lookup.response.thread-pool.size' =
'4'"
+ + ")";
+
+ tEnv.executeSql(sourceTable);
+ tEnv.executeSql(sourceView);
+ tEnv.executeSql(lookupTable);
+
+ // WHEN - Join on url field
+ String joinQuery =
+ "SELECT o.id, o.url, c.name, c.company, c.email FROM
OrdersWithUrl AS o"
+ + " JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time
AS c"
+ + " ON c.url = o.url";
+
+ TableResult result = tEnv.executeSql(joinQuery);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
+
+ // THEN
+ SortedSet<Row> collectedRows = getCollectedRows(result);
+
+ // Should have 3 rows with url populated from source table
+ assertThat(collectedRows.size()).isEqualTo(3);
+
+ // Verify that url field is populated in results
+ for (Row row : collectedRows) {
+ assertThat(row.getField(1)).isNotNull(); // url should not be null
+ assertThat(row.getField(2)).isNotNull(); // name from HTTP response
+ assertThat(row.getField(3)).isNotNull(); // company from HTTP
response
+ assertThat(row.getField(4)).isNotNull(); // email from HTTP
response
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testHttpLookupJoinWithCache(boolean isAsync) throws Exception {
@@ -1002,7 +1095,14 @@ class HttpLookupTableSourceITCaseTest {
+ "AND o.id2 = c.id2";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
+
+ // Wait for all async HTTP requests to complete and be processed
+ await().atMost(2, SECONDS)
+ .untilAsserted(
+ () ->
+ assertThat(wireMockServer.getAllServeEvents())
+ .hasSizeGreaterThanOrEqualTo(maxRows));
// THEN
return getCollectedRows(result);
@@ -1043,7 +1143,7 @@ class HttpLookupTableSourceITCaseTest {
+ "'asyncPolling' = 'true',"
+ "'lookup-request.format' = 'json',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
- + "'table.exec.async-lookup.timeout' = '120s',"
+ + "'table.exec.async-lookup.timeout' = '20s',"
// Template creates flat structure: {"id": {{id}},
"id2": {{id2}}}
// This proves the template feature works (unit tests
cover nested cases)
+ "'http.request.body-template' = '{\"id\": {{id}},
\"id2\": {{id2}}}'"
@@ -1140,7 +1240,7 @@ class HttpLookupTableSourceITCaseTest {
+ "'asyncPolling' = 'true',"
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
- + "'table.exec.async-lookup.timeout' = '120s',"
+ + "'table.exec.async-lookup.timeout' = '20s',"
// Map "id" column to "customer" query parameter
+ "'http.request.query-param-fields-with-key' =
'id:customer',"
+ "'http.request.query-param-fields' = 'id2'"
@@ -1158,7 +1258,7 @@ class HttpLookupTableSourceITCaseTest {
+ "AND o.id2 = c.id2";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -1232,7 +1332,7 @@ class HttpLookupTableSourceITCaseTest {
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'lookup-request.format' = 'json',"
+ "'table.exec.async-lookup.buffer-capacity' = '50',"
- + "'table.exec.async-lookup.timeout' = '120s',"
+ + "'table.exec.async-lookup.timeout' = '20s',"
// Template maps body_customer to "customer" in
request body
+ "'http.request.body-template' = '{\"customer\":
{{body_customer}}, \"id2\": {{id2}}}'"
+ ")";
@@ -1249,7 +1349,7 @@ class HttpLookupTableSourceITCaseTest {
+ "AND o.id2 = c.id2";
TableResult result = tEnv.executeSql(joinQuery);
- result.await(15, TimeUnit.SECONDS);
+ result.await(SECONDS_TO_WAIT_FOR_RESPONSE, TimeUnit.SECONDS);
// THEN
SortedSet<Row> collectedRows = getCollectedRows(result);
@@ -1290,7 +1390,7 @@ class HttpLookupTableSourceITCaseTest {
+ " proc_time AS PROCTIME()"
+ ") WITH ("
+ "'connector' = 'datagen',"
- + "'rows-per-second' = '1',"
+ + "'rows-per-second' = '100',"
+ "'fields.id.kind' = 'sequence',"
+ "'fields.id.start' = '1',"
+ "'fields.id.end' = '"
@@ -1818,7 +1918,7 @@ class HttpLookupTableSourceITCaseTest {
.append(spec.useAsync ? "true" : "false")
.append("',")
.append("'table.exec.async-lookup.buffer-capacity' = '50',")
- .append("'table.exec.async-lookup.timeout' = '120s'")
+ .append("'table.exec.async-lookup.timeout' = '20s'")
.append(")");
return sql.toString();
}
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 a03b452..a610b56 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
@@ -197,7 +197,8 @@ class HttpLookupTableSourceTest {
(HttpLookupTableSource) createTableSource(SCHEMA,
getOptions());
LookupTableSource.LookupRuntimeProvider lookupProvider =
- tableSource.getLookupRuntimeProvider(new
LookupRuntimeProviderContext(lookupKey));
+ tableSource.getLookupRuntimeProvider(
+ new LookupRuntimeProviderContext(lookupKey, false));
HttpTableLookupFunction tableFunction =
(HttpTableLookupFunction)
((LookupFunctionProvider)
lookupProvider).createLookupFunction();
@@ -230,7 +231,7 @@ class HttpLookupTableSourceTest {
AsyncLookupFunctionProvider lookupProvider =
(AsyncLookupFunctionProvider)
tableSource.getLookupRuntimeProvider(
- new LookupRuntimeProviderContext(lookupKey));
+ new LookupRuntimeProviderContext(lookupKey,
false));
AsyncHttpTableLookupFunction tableFunction =
(AsyncHttpTableLookupFunction)
lookupProvider.createAsyncLookupFunction();
@@ -331,7 +332,7 @@ class HttpLookupTableSourceTest {
new HttpLookupTableSource(null, options, null, null, cache);
int[][] lookupKeys = {{1, 2}};
LookupTableSource.LookupContext lookupContext =
- new LookupRuntimeProviderContext(lookupKeys);
+ new LookupRuntimeProviderContext(lookupKeys, false);
return tableSource.getLookupRuntimeProvider(null, null, null);
}
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
new file mode 100644
index 0000000..f078bd6
--- /dev/null
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunctionPopulateJoinKeysTest.java
@@ -0,0 +1,347 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.http.table.lookup;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.connector.http.clients.PollingClient;
+import org.apache.flink.connector.http.clients.PollingClientFactory;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Unit tests for {@link HttpTableLookupFunction} populateNullJoinKeys
method. */
+@ExtendWith(MockitoExtension.class)
+class HttpTableLookupFunctionPopulateJoinKeysTest {
+
+ @Mock private PollingClientFactory pollingClientFactory;
+
+ @Mock private PollingClient pollingClient;
+
+ @Mock private DeserializationSchema<RowData> responseSchemaDecoder;
+
+ private HttpLookupConfig httpLookupConfig;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ httpLookupConfig = HttpLookupConfig.builder().build();
+ when(pollingClientFactory.createPollClient(any(),
any())).thenReturn(pollingClient);
+
+ // Mock pollingClient.open to do nothing
+ Mockito.doNothing().when(pollingClient).open(any());
+ }
+
+ private FunctionContext createMockFunctionContext() {
+ FunctionContext context = mock(FunctionContext.class);
+ MetricGroup metricGroup = mock(MetricGroup.class);
+ when(context.getMetricGroup()).thenReturn(metricGroup);
+ when(metricGroup.gauge(anyString(), any())).thenReturn(null);
+ return context;
+ }
+
+ @Test
+ void testPopulateNullJoinKeys_singleJoinKey() throws Exception {
+ // Setup: Create a schema with 3 fields: url (join key), name, value
+ DataType producedDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("url", DataTypes.STRING()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.STRING()));
+
+ // Create lookup row with single join key "url"
+ LookupRow lookupRow = new LookupRow();
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "url", RowData.createFieldGetter(new VarCharType(),
0)));
+ lookupRow.setLookupPhysicalRowDataType(producedDataType);
+
+ // Create the lookup function
+ HttpTableLookupFunction lookupFunction =
+ new HttpTableLookupFunction(
+ pollingClientFactory,
+ responseSchemaDecoder,
+ lookupRow,
+ httpLookupConfig,
+ new MetadataConverter[0],
+ null,
+ producedDataType);
+
+ // Create keyRow with join key value
+ GenericRowData keyRow = new GenericRowData(1);
+ keyRow.setField(0, StringData.fromString("http://example.com";));
+
+ // Create HTTP response row with null join key but populated other
fields
+ GenericRowData httpResponseRow = new GenericRowData(3);
+ httpResponseRow.setField(0, null); // url is null
+ httpResponseRow.setField(1, StringData.fromString("test-name"));
+ httpResponseRow.setField(2, StringData.fromString("test-value"));
+
+ // Mock the HTTP client to return the response
+ HttpRowDataWrapper wrapper =
+ HttpRowDataWrapper.builder()
+ .data(Collections.singletonList(httpResponseRow))
+ .httpCompletionState(HttpCompletionState.SUCCESS)
+ .httpStatusCode(200)
+ .build();
+ when(pollingClient.pull(any())).thenReturn(wrapper);
+
+ // Open the function
+ FunctionContext context = createMockFunctionContext();
+ lookupFunction.open(context);
+
+ // Execute lookup
+ Collection<RowData> result = lookupFunction.lookup(keyRow);
+
+ // Verify: The result should have the join key populated
+ assertThat(result).hasSize(1);
+ RowData resultRow = result.iterator().next();
+ assertThat(resultRow.getArity()).isEqualTo(3);
+
assertThat(resultRow.getString(0)).isEqualTo(StringData.fromString("http://example.com";));
+
assertThat(resultRow.getString(1)).isEqualTo(StringData.fromString("test-name"));
+
assertThat(resultRow.getString(2)).isEqualTo(StringData.fromString("test-value"));
+ }
+
+ @Test
+ void testPopulateNullJoinKeys_threeJoinKeys() throws Exception {
+ // Setup: Create a schema with 5 fields: id, category, region (all
join keys), name, value
+ DataType producedDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.STRING()),
+ DataTypes.FIELD("category", DataTypes.STRING()),
+ DataTypes.FIELD("region", DataTypes.STRING()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.STRING()));
+
+ // Create lookup row with three join keys
+ LookupRow lookupRow = new LookupRow();
+ LogicalType stringType = new VarCharType();
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "id", RowData.createFieldGetter(stringType, 0)));
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "category", RowData.createFieldGetter(stringType, 1)));
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "region", RowData.createFieldGetter(stringType, 2)));
+ lookupRow.setLookupPhysicalRowDataType(producedDataType);
+
+ // Create the lookup function
+ HttpTableLookupFunction lookupFunction =
+ new HttpTableLookupFunction(
+ pollingClientFactory,
+ responseSchemaDecoder,
+ lookupRow,
+ httpLookupConfig,
+ new MetadataConverter[0],
+ null,
+ producedDataType);
+
+ // Create keyRow with three join key values
+ GenericRowData keyRow = new GenericRowData(3);
+ keyRow.setField(0, StringData.fromString("id-123"));
+ keyRow.setField(1, StringData.fromString("electronics"));
+ keyRow.setField(2, StringData.fromString("us-west"));
+
+ // Create HTTP response row with null join keys but populated other
fields
+ GenericRowData httpResponseRow = new GenericRowData(5);
+ httpResponseRow.setField(0, null); // id is null
+ httpResponseRow.setField(1, null); // category is null
+ httpResponseRow.setField(2, null); // region is null
+ httpResponseRow.setField(3, StringData.fromString("Product Name"));
+ httpResponseRow.setField(4, StringData.fromString("Product Value"));
+
+ // Mock the HTTP client to return the response
+ HttpRowDataWrapper wrapper =
+ HttpRowDataWrapper.builder()
+ .data(Collections.singletonList(httpResponseRow))
+ .httpCompletionState(HttpCompletionState.SUCCESS)
+ .httpStatusCode(200)
+ .build();
+ when(pollingClient.pull(any())).thenReturn(wrapper);
+
+ // Open the function
+ FunctionContext context = createMockFunctionContext();
+ lookupFunction.open(context);
+
+ // Execute lookup
+ Collection<RowData> result = lookupFunction.lookup(keyRow);
+
+ // Verify: All three join keys should be populated
+ assertThat(result).hasSize(1);
+ RowData resultRow = result.iterator().next();
+ assertThat(resultRow.getArity()).isEqualTo(5);
+
assertThat(resultRow.getString(0)).isEqualTo(StringData.fromString("id-123"));
+
assertThat(resultRow.getString(1)).isEqualTo(StringData.fromString("electronics"));
+
assertThat(resultRow.getString(2)).isEqualTo(StringData.fromString("us-west"));
+
assertThat(resultRow.getString(3)).isEqualTo(StringData.fromString("Product
Name"));
+
assertThat(resultRow.getString(4)).isEqualTo(StringData.fromString("Product
Value"));
+ }
+
+ @Test
+ void testPopulateNullJoinKeys_nullNonKeyFieldsRemainNull() throws
Exception {
+ // Setup: Create a schema with 5 fields: url (join key), name, value,
description, status
+ DataType producedDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("url", DataTypes.STRING()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.STRING()),
+ DataTypes.FIELD("description", DataTypes.STRING()),
+ DataTypes.FIELD("status", DataTypes.STRING()));
+
+ // Create lookup row with single join key "url"
+ LookupRow lookupRow = new LookupRow();
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "url", RowData.createFieldGetter(new VarCharType(),
0)));
+ lookupRow.setLookupPhysicalRowDataType(producedDataType);
+
+ // Create the lookup function
+ HttpTableLookupFunction lookupFunction =
+ new HttpTableLookupFunction(
+ pollingClientFactory,
+ responseSchemaDecoder,
+ lookupRow,
+ httpLookupConfig,
+ new MetadataConverter[0],
+ null,
+ producedDataType);
+
+ // Create keyRow with join key value
+ GenericRowData keyRow = new GenericRowData(1);
+ keyRow.setField(0, StringData.fromString("http://example.com";));
+
+ // Create HTTP response row with null join key and some null non-key
fields
+ GenericRowData httpResponseRow = new GenericRowData(5);
+ httpResponseRow.setField(0, null); // url is null (join key - will be
populated)
+ httpResponseRow.setField(1, StringData.fromString("test-name"));
+ httpResponseRow.setField(2, null); // value is null (non-key - should
remain null)
+ httpResponseRow.setField(3, StringData.fromString("test-description"));
+ httpResponseRow.setField(4, null); // status is null (non-key - should
remain null)
+
+ // Mock the HTTP client to return the response
+ HttpRowDataWrapper wrapper =
+ HttpRowDataWrapper.builder()
+ .data(Collections.singletonList(httpResponseRow))
+ .httpCompletionState(HttpCompletionState.SUCCESS)
+ .httpStatusCode(200)
+ .build();
+ when(pollingClient.pull(any())).thenReturn(wrapper);
+
+ // Open the function
+ FunctionContext context = createMockFunctionContext();
+ lookupFunction.open(context);
+
+ // Execute lookup
+ Collection<RowData> result = lookupFunction.lookup(keyRow);
+
+ // Verify: Join key populated, non-key nulls remain null
+ assertThat(result).hasSize(1);
+ RowData resultRow = result.iterator().next();
+ assertThat(resultRow.getArity()).isEqualTo(5);
+
assertThat(resultRow.getString(0)).isEqualTo(StringData.fromString("http://example.com";));
+
assertThat(resultRow.getString(1)).isEqualTo(StringData.fromString("test-name"));
+ assertThat(resultRow.isNullAt(2)).isTrue(); // value should remain null
+
assertThat(resultRow.getString(3)).isEqualTo(StringData.fromString("test-description"));
+ assertThat(resultRow.isNullAt(4)).isTrue(); // status should remain
null
+ }
+
+ @Test
+ void testPopulateNullJoinKeys_nonNullFieldNotOverwritten() throws
Exception {
+ // Setup: Create a schema with 3 fields: url (join key), name, value
+ DataType producedDataType =
+ DataTypes.ROW(
+ DataTypes.FIELD("url", DataTypes.STRING()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.STRING()));
+
+ // Create lookup row with single join key "url"
+ LookupRow lookupRow = new LookupRow();
+ lookupRow.addLookupEntry(
+ new RowDataSingleValueLookupSchemaEntry(
+ "url", RowData.createFieldGetter(new VarCharType(),
0)));
+ lookupRow.setLookupPhysicalRowDataType(producedDataType);
+
+ // Create the lookup function
+ HttpTableLookupFunction lookupFunction =
+ new HttpTableLookupFunction(
+ pollingClientFactory,
+ responseSchemaDecoder,
+ lookupRow,
+ httpLookupConfig,
+ new MetadataConverter[0],
+ null,
+ producedDataType);
+
+ // Create keyRow with join key value
+ GenericRowData keyRow = new GenericRowData(1);
+ keyRow.setField(0, StringData.fromString("http://example.com";));
+
+ // Create HTTP response row with NON-NULL join key (should not be
overwritten)
+ GenericRowData httpResponseRow = new GenericRowData(3);
+ httpResponseRow.setField(0,
StringData.fromString("http://response-url.com";));
+ httpResponseRow.setField(1, StringData.fromString("test-name"));
+ httpResponseRow.setField(2, StringData.fromString("test-value"));
+
+ // Mock the HTTP client to return the response
+ HttpRowDataWrapper wrapper =
+ HttpRowDataWrapper.builder()
+ .data(Collections.singletonList(httpResponseRow))
+ .httpCompletionState(HttpCompletionState.SUCCESS)
+ .httpStatusCode(200)
+ .build();
+ when(pollingClient.pull(any())).thenReturn(wrapper);
+
+ // Open the function
+ FunctionContext context = createMockFunctionContext();
+ lookupFunction.open(context);
+
+ // Execute lookup
+ Collection<RowData> result = lookupFunction.lookup(keyRow);
+
+ // Verify: The non-null url from response should NOT be overwritten
+ assertThat(result).hasSize(1);
+ RowData resultRow = result.iterator().next();
+ assertThat(resultRow.getString(0))
+ .isEqualTo(StringData.fromString("http://response-url.com";));
+
assertThat(resultRow.getString(1)).isEqualTo(StringData.fromString("test-name"));
+
assertThat(resultRow.getString(2)).isEqualTo(StringData.fromString("test-value"));
+ }
+}
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientConnectionTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientConnectionTest.java
index f0ee628..fbeacc9 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientConnectionTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientConnectionTest.java
@@ -125,7 +125,7 @@ class JavaNetHttpPollingClientConnectionTest {
@BeforeEach
void setUp() {
int[][] lookupKey = {{}};
- this.dynamicTableSourceContext = new
LookupRuntimeProviderContext(lookupKey);
+ this.dynamicTableSourceContext = new
LookupRuntimeProviderContext(lookupKey, false);
this.lookupRowData =
GenericRowData.of(StringData.fromString("1"),
StringData.fromString("2"));
@@ -292,8 +292,8 @@ class JavaNetHttpPollingClientConnectionTest {
// GIVEN
this.stubMapping = setUpServerStub(201);
- configuration.setString(SOURCE_LOOKUP_HTTP_SUCCESS_CODES,
successCodesExpression);
- configuration.setString(
+ configuration.set(SOURCE_LOOKUP_HTTP_SUCCESS_CODES,
successCodesExpression);
+ configuration.set(
SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
ignoredResponseCodesExpression);
JavaNetHttpPollingClient pollingClient = setUpPollingClient();
@@ -545,10 +545,9 @@ class JavaNetHttpPollingClientConnectionTest {
void shouldSetIgnoreStatusCodeCompletionStateForIgnoredStatusCodes()
throws ConfigurationException {
// GIVEN - Configure client with ignored status codes (404, 503)
- configuration.setString(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503");
- configuration.setString(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
- configuration.setString(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES.key(), "");
+ configuration.set(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503");
+ configuration.set(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
+
configuration.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES,
"");
// Set up WireMock to return 404
this.stubMapping =
@@ -572,10 +571,9 @@ class JavaNetHttpPollingClientConnectionTest {
@Test
void shouldSetIgnoreStatusCodeForMultipleIgnoredCodes() throws
ConfigurationException {
// GIVEN - Configure client with multiple ignored status codes
- configuration.setString(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503,429");
- configuration.setString(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
- configuration.setString(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES.key(), "");
+ configuration.set(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503,429");
+ configuration.set(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
+
configuration.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES,
"");
// Set up WireMock to return 503
this.stubMapping =
@@ -602,10 +600,9 @@ class JavaNetHttpPollingClientConnectionTest {
@Test
void shouldNotSetIgnoreStatusCodeForNonIgnoredCodes() throws
ConfigurationException {
// GIVEN - Configure client with ignored status codes (404, 503)
- configuration.setString(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503");
- configuration.setString(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
- configuration.setString(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES.key(), "");
+ configuration.set(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404,503");
+ configuration.set(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
+
configuration.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES,
"");
// Set up WireMock to return 200 (success, not ignored)
this.stubMapping = setUpServerStub(200);
@@ -624,10 +621,9 @@ class JavaNetHttpPollingClientConnectionTest {
@Test
void shouldReturnMetadataForIgnoredStatusCode() throws
ConfigurationException {
// GIVEN - Configure client with ignored status codes (404)
- configuration.setString(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES,
"404");
- configuration.setString(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
- configuration.setString(
-
HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES.key(), "");
+ configuration.set(SOURCE_LOOKUP_HTTP_IGNORED_RESPONSE_CODES, "404");
+ configuration.set(SOURCE_LOOKUP_HTTP_SUCCESS_CODES, "200");
+
configuration.set(HttpLookupConnectorOptions.SOURCE_LOOKUP_HTTP_RETRY_CODES,
"");
// Set up WireMock to return 404 with custom headers
this.stubMapping =
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientHttpsConnectionTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientHttpsConnectionTest.java
index 3f4e1a9..dd03e24 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientHttpsConnectionTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientHttpsConnectionTest.java
@@ -90,7 +90,7 @@ public class JavaNetHttpPollingClientHttpsConnectionTest
extends HttpsConnection
super.setUp();
httpServerPort = WireMockServerPortAllocator.getSecureServerPort();
int[][] lookupKey = {{0, 1}};
- this.dynamicTableSourceContext = new
LookupRuntimeProviderContext(lookupKey);
+ this.dynamicTableSourceContext = new
LookupRuntimeProviderContext(lookupKey, false);
this.lookupRowData =
GenericRowData.of(StringData.fromString("1"),
StringData.fromString("2"));
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientWithWireTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientWithWireTest.java
index 33421c2..d361185 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientWithWireTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClientWithWireTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.connector.http.table.lookup;
import org.apache.flink.api.common.RuntimeExecutionMode;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.connector.http.WireMockServerPortAllocator;
@@ -94,7 +93,6 @@ public class JavaNetHttpPollingClientWithWireTest {
wireMockServer.start();
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- env.setRestartStrategy(RestartStrategies.noRestart());
Configuration config = new Configuration();
config.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.STREAMING);
env.configure(config, getClass().getClassLoader());
@@ -153,10 +151,10 @@ public class JavaNetHttpPollingClientWithWireTest {
HttpRequest newHttpRequest =
client.updateHttpRequestIfRequired(request,
oidcHeaderPreProcessor);
assertThat(httpRequest).isEqualTo(newHttpRequest);
- configuration.setString(
- SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL.key(),
+ configuration.set(
+ SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL,
"http://localhost:"; + SERVER_PORT + "/auth");
- configuration.setString(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
BEARER_REQUEST);
+ configuration.set(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST,
BEARER_REQUEST);
configuration.set(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION,
Duration.ofSeconds(1L));
client =
new JavaNetHttpPollingClient(
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
index 31d28d1..d592d8d 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonAndUrlQueryCreatorFactoryTest.java
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.Test;
import java.util.List;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.LOOKUP_REQUEST_FORMAT;
import static
org.apache.flink.connector.http.table.lookup.HttpLookupTableSourceFactory.row;
import static
org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_BODY_TEMPLATE;
import static
org.apache.flink.connector.http.table.lookup.querycreators.GenericJsonAndUrlQueryCreatorFactory.REQUEST_QUERY_PARAM_FIELDS;
@@ -70,7 +71,7 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
+ "was called before this test execution.")
.isFalse();
- this.config.setString("lookup-request.format",
CustomJsonFormatFactory.IDENTIFIER);
+ this.config.set(LOOKUP_REQUEST_FORMAT,
CustomJsonFormatFactory.IDENTIFIER);
this.config.setString(
String.format(
"lookup-request.format.%s.%s",
@@ -97,7 +98,7 @@ class GenericJsonAndUrlQueryCreatorFactoryTest {
}
private void createUsingFactory(boolean async) {
- this.config.setBoolean(HttpLookupConnectorOptions.ASYNC_POLLING,
async);
+ this.config.set(HttpLookupConnectorOptions.ASYNC_POLLING, async);
LookupRow lookupRow =
new LookupRow()
.addLookupEntry(
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactoryTest.java
index 4cf4c3f..7bd1edf 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/GenericJsonQueryCreatorFactoryTest.java
@@ -38,6 +38,7 @@ import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.List;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.LOOKUP_REQUEST_FORMAT;
import static
org.apache.flink.connector.http.table.lookup.HttpLookupTableSourceFactory.row;
import static org.assertj.core.api.Assertions.assertThat;
@@ -73,13 +74,13 @@ class GenericJsonQueryCreatorFactoryTest {
new FactoryUtil.DefaultDynamicTableContext(
ObjectIdentifier.of("default", "default", "test"),
new ResolvedCatalogTable(
- CatalogTable.of(
- Schema.newBuilder()
-
.fromResolvedSchema(resolvedSchema)
- .build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+
.fromResolvedSchema(resolvedSchema)
+ .build())
+ .options(Collections.emptyMap())
+ .build(),
resolvedSchema),
Collections.emptyMap(),
config,
@@ -96,7 +97,7 @@ class GenericJsonQueryCreatorFactoryTest {
+ "was called before this test execution.")
.isFalse();
- this.config.setString("lookup-request.format",
CustomFormatFactory.IDENTIFIER);
+ this.config.set(LOOKUP_REQUEST_FORMAT, CustomFormatFactory.IDENTIFIER);
this.config.setString(
String.format(
"lookup-request.format.%s.%s",
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/QueryCreatorUtils.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/QueryCreatorUtils.java
index f985df1..6701ace 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/QueryCreatorUtils.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/lookup/querycreators/QueryCreatorUtils.java
@@ -42,11 +42,13 @@ public class QueryCreatorUtils {
return new FactoryUtil.DefaultDynamicTableContext(
ObjectIdentifier.of("default", "default", "test"),
new ResolvedCatalogTable(
- CatalogTable.of(
-
Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(),
- null,
- Collections.emptyList(),
- Collections.emptyMap()),
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+
.fromResolvedSchema(resolvedSchema)
+ .build())
+ .options(Collections.emptyMap())
+ .build(),
resolvedSchema),
Collections.emptyMap(),
config,
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java
index 7b71223..bcb1644 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/HttpHeaderUtilsTest.java
@@ -38,8 +38,8 @@ public class HttpHeaderUtilsTest {
HeaderPreprocessor headerPreprocessor =
HttpHeaderUtils.createOIDCHeaderPreprocessor(configuration);
assertThat(headerPreprocessor).isNull();
-
configuration.setString(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL.key(),
"http://aaa";);
- configuration.setString(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST.key(),
"ccc");
+ configuration.set(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_ENDPOINT_URL,
"http://aaa";);
+ configuration.set(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_REQUEST, "ccc");
configuration.set(SOURCE_LOOKUP_OIDC_AUTH_TOKEN_EXPIRY_REDUCTION,
Duration.ofSeconds(1));
headerPreprocessor =
HttpHeaderUtils.createOIDCHeaderPreprocessor(configuration);
assertThat(headerPreprocessor).isNotNull();
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java
index 01cd3c1..2c671e6 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/utils/JavaNetHttpClientFactoryTest.java
@@ -36,10 +36,10 @@ import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import static
org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SOURCE_PROXY_HOST;
-import static
org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SOURCE_PROXY_PASSWORD;
-import static
org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SOURCE_PROXY_PORT;
-import static
org.apache.flink.connector.http.config.HttpConnectorConfigConstants.SOURCE_PROXY_USERNAME;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_HOST;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_PASSWORD;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_PORT;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_PROXY_USERNAME;
import static org.assertj.core.api.Assertions.assertThat;
class JavaNetHttpClientFactoryTest {
@@ -50,10 +50,10 @@ class JavaNetHttpClientFactoryTest {
public void shouldGetClientWithAuthenticator() throws UnknownHostException
{
Properties properties = new Properties();
Configuration configuration = new Configuration();
- configuration.setString(SOURCE_PROXY_HOST, "google");
- configuration.setString(SOURCE_PROXY_PORT,
String.valueOf(PROXY_SERVER_PORT));
- configuration.setString(SOURCE_PROXY_USERNAME, "username");
- configuration.setString(SOURCE_PROXY_PASSWORD, "password");
+ configuration.set(SOURCE_LOOKUP_PROXY_HOST, "google");
+ configuration.set(SOURCE_LOOKUP_PROXY_PORT, PROXY_SERVER_PORT);
+ configuration.set(SOURCE_LOOKUP_PROXY_USERNAME, "username");
+ configuration.set(SOURCE_LOOKUP_PROXY_PASSWORD, "password");
HttpLookupConfig lookupConfig =
HttpLookupConfig.builder()
@@ -90,8 +90,8 @@ class JavaNetHttpClientFactoryTest {
public void shouldGetClientWithoutAuthenticator() throws
UnknownHostException {
Properties properties = new Properties();
Configuration configuration = new Configuration();
- configuration.setString(SOURCE_PROXY_HOST, "google");
- configuration.setString(SOURCE_PROXY_PORT,
String.valueOf(PROXY_SERVER_PORT));
+ configuration.set(SOURCE_LOOKUP_PROXY_HOST, "google");
+ configuration.set(SOURCE_LOOKUP_PROXY_PORT, PROXY_SERVER_PORT);
HttpLookupConfig lookupConfig =
HttpLookupConfig.builder()
diff --git a/pom.xml b/pom.xml
index 45c6b29..818610e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -23,7 +23,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-parent</artifactId>
- <version>1.1.0</version>
+ <version>2.0.0</version>
</parent>
<modules>
<module>flink-connector-http</module>
@@ -62,7 +62,7 @@ under the License.
<!-- IMPORTANT: If you update Flink, remember to update link to its
docs in maven-javadoc-plugin <links>
section, omitting the patch part (so for 1.15.0 use 1.15).
TODO is this still valid in Flink connector. -->
- <flink.version>1.20.0</flink.version>
+ <flink.version>2.2.0</flink.version>
<flink.shaded.jackson.version>2.18.2</flink.shaded.jackson.version>
<flink.shaded.version>20.0</flink.shaded.version>
<flink.parent.artifactId>flink-connector-http-parent</flink.parent.artifactId>
@@ -76,6 +76,7 @@ under the License.
<!-- 3rd party versions -->
<assertj.core.version>3.27.7</assertj.core.version>
+ <awaitility.version>4.2.0</awaitility.version>
<!-- explicitly include a java 21 supporting version of byte buddy -->
<bytebuddy.version>1.14.17</bytebuddy.version>
<guava.version>27.0-jre</guava.version>
@@ -84,10 +85,9 @@ under the License.
<json-smart.version>2.5.2</json-smart.version>
<json-unit-core.version>2.40.1</json-unit-core.version>
<jsr305.version>1.3.9</jsr305.version>
- <junit5.version>5.10.1</junit5.version>
+ <junit5.version>5.11.4</junit5.version>
<junit.jupiter.version>${junit5.version}</junit.jupiter.version>
<lombok.version>1.18.38</lombok.version>
- <!-- TODO remove mockito dependency 4.6.1 use level that supports java
21-->
<mockito.version>5.2.0</mockito.version>
<mockito-inline.version>5.2.0</mockito-inline.version>
<resilence4j.version>1.7.1</resilence4j.version>
@@ -138,19 +138,6 @@ under the License.
<dependencies>
<!-- Apache Flink dependencies -->
<!-- These dependencies are provided, because they should not be
packaged into the JAR file. -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java</artifactId>
- <version>${flink.version}</version>
- <exclusions>
- <!--exclusion required to pass convergence check in CI -->
- <exclusion>
- <groupId>com.esotericsoftware.kryo</groupId>
- <artifactId>kryo</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-annotations</artifactId>
@@ -529,6 +516,13 @@ under the License.
</exclusion>
</exclusions>
</dependency>
+
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
@@ -650,6 +644,9 @@ under the License.
<ignoredDependencies>
<!--These dependencies are required for the
unit tests to succeed -->
<ignoredDependency>org.mockito:mockito-inline:jar:4.6.1</ignoredDependency>
+ <!-- Runtime dependencies loaded via
ServiceLoader for test execution -->
+
<ignoredDependency>org.apache.flink:flink-clients:jar:*:test</ignoredDependency>
+
<ignoredDependency>org.apache.flink:flink-streaming-java:jar:*:test</ignoredDependency>
</ignoredDependencies>
</configuration>
</execution>
