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 8b98b78 [FLINK-40072][connector-http] Read request.timeout as the
declared Duration option (#52)
8b98b78 is described below
commit 8b98b78137655c970e4e16d597b48c8168552001
Author: Bharath Reddy Gunapati
<[email protected]>
AuthorDate: Mon Jul 20 18:13:08 2026 +0530
[FLINK-40072][connector-http] Read request.timeout as the declared Duration
option (#52)
* [FLINK-40072][connector-http] Read request.timeout as the declared
Duration option
request.timeout is declared as a ConfigOption<Duration> for both the lookup
source (SOURCE_LOOKUP_REQUEST_TIMEOUT) and the sink (SINK_REQUEST_TIMEOUT)
with a
default of 30s, and the docs describe it as a Flink Duration. However, the
runtime
never read the declared option: it parsed the raw config string with
Integer.parseInt(...), so any Duration-formatted value such as '30s' threw
NumberFormatException at query time. The failure is deferred because the
http.*
namespace is skipped by FactoryUtil#validateExcept(...), so CREATE TABLE
succeeds
and only the first request crashes.
This wires the runtime to the declared options:
- Lookup: RequestFactoryBase reads SOURCE_LOOKUP_REQUEST_TIMEOUT from
ReadableConfig as a Duration and applies it directly.
- Sink: AbstractRequestSubmitter parses the value with
TimeUtils.parseDuration(...);
PerRequestSubmitter and BatchRequestSubmitter use the Duration directly.
- Removes the now-unused integer-seconds field and
DEFAULT_REQUEST_TIMEOUT_SECONDS
constant; the default comes from the option (Duration.ofSeconds(30)).
- Updates docs to describe the value as a Flink Duration ('30s', '1min').
Behavioral note: values are now parsed with standard Flink Duration
semantics
(consistent with the sibling connection.timeout option). Unit-suffixed
values now
work; a bare number such as '30' now means 30ms rather than 30s, so bare
integers
must be rewritten with a unit (e.g. '30s'). The defect was introduced by
FLINK-39364 (#31).
Tests: end-to-end lookup join with '30s' (HttpLookupTableSourceITCaseTest),
Duration parsing (BodyBasedRequestFactoryTest, BatchRequestSubmitterTest),
and
sink option acceptance (HttpDynamicTableSinkFactoryTest).
* [FLINK-40072][connector-http] Address review feedback on request.timeout
PR
Clarify sink Properties vs ReadableConfig asymmetry, tighten test assertions
and comments per maintainer review, and fix requestTimeoutOptionTest to
handle
successful INSERT without failing on a null throwable.
* [FLINK-40072][connector-http] Address review feedback on request.timeout
PR
---
docs/content.zh/docs/connectors/datastream/http.md | 2 +-
docs/content/docs/connectors/datastream/http.md | 2 +-
.../sink/httpclient/AbstractRequestSubmitter.java | 19 +++++---
.../sink/httpclient/BatchRequestSubmitter.java | 3 +-
.../http/sink/httpclient/PerRequestSubmitter.java | 3 +-
.../http/table/lookup/RequestFactoryBase.java | 31 ++++++-------
.../sink/httpclient/BatchRequestSubmitterTest.java | 32 +++++++++++++
.../table/lookup/BodyBasedRequestFactoryTest.java | 54 ++++++++++++++++++++++
.../lookup/HttpLookupTableSourceITCaseTest.java | 44 +++++++++++++++++-
.../sink/HttpDynamicTableSinkFactoryTest.java | 15 ++++--
10 files changed, 169 insertions(+), 36 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/http.md
b/docs/content.zh/docs/connectors/datastream/http.md
index e1e6cfa..c3a959d 100644
--- a/docs/content.zh/docs/connectors/datastream/http.md
+++ b/docs/content.zh/docs/connectors/datastream/http.md
@@ -68,7 +68,7 @@ These options are specified on the builder using the
setProperty method.
| flink.connector.http.security.cert.client | optional | Path
to trusted certificate that should be used by connector's HTTP client for mTLS
communication.
|
| flink.connector.http.security.key.client | optional | Path
to trusted private key that should be used by connector's HTTP client for mTLS
communication.
|
| flink.connector.http.security.cert.server.allowSelfSigned | optional |
Accept untrusted certificates for TLS communication.
|
-| flink.connector.http.sink.request.timeout | optional | Sets
HTTP request timeout in seconds. If not specified, the default value of 30
seconds will be used.
|
+| flink.connector.http.sink.request.timeout | optional | Sets
HTTP request timeout for the HTTP sink as a Duration (e.g. `30s`, `1min`). If
not specified, the default value of `30s` will be used.
|
| flink.connector.http.sink.writer.thread-pool.size | optional | Sets
the size of pool thread for HTTP Sink request processing. Increasing this value
would mean that more concurrent requests can be processed in the same time. If
not specified, the default value of 1 thread will be used. |
| flink.connector.http.sink.writer.request.mode | optional | Sets
the Http Sink request submission mode. Two modes are available: `single` and
`batch`. Defaults to `batch` if not specified. |
| flink.connector.http.sink.request.batch.size | optional |
Applicable only for `flink.connector.http.sink.writer.request.mode = batch`.
Sets number of individual events/requests that will be submitted as one HTTP
request by HTTP sink. The default value is 500 which is same as HTTP Sink
`maxBatchSize` |
diff --git a/docs/content/docs/connectors/datastream/http.md
b/docs/content/docs/connectors/datastream/http.md
index e1e6cfa..c3a959d 100644
--- a/docs/content/docs/connectors/datastream/http.md
+++ b/docs/content/docs/connectors/datastream/http.md
@@ -68,7 +68,7 @@ These options are specified on the builder using the
setProperty method.
| flink.connector.http.security.cert.client | optional | Path
to trusted certificate that should be used by connector's HTTP client for mTLS
communication.
|
| flink.connector.http.security.key.client | optional | Path
to trusted private key that should be used by connector's HTTP client for mTLS
communication.
|
| flink.connector.http.security.cert.server.allowSelfSigned | optional |
Accept untrusted certificates for TLS communication.
|
-| flink.connector.http.sink.request.timeout | optional | Sets
HTTP request timeout in seconds. If not specified, the default value of 30
seconds will be used.
|
+| flink.connector.http.sink.request.timeout | optional | Sets
HTTP request timeout for the HTTP sink as a Duration (e.g. `30s`, `1min`). If
not specified, the default value of `30s` will be used.
|
| flink.connector.http.sink.writer.thread-pool.size | optional | Sets
the size of pool thread for HTTP Sink request processing. Increasing this value
would mean that more concurrent requests can be processed in the same time. If
not specified, the default value of 1 thread will be used. |
| flink.connector.http.sink.writer.request.mode | optional | Sets
the Http Sink request submission mode. Two modes are available: `single` and
`batch`. Defaults to `batch` if not specified. |
| flink.connector.http.sink.request.batch.size | optional |
Applicable only for `flink.connector.http.sink.writer.request.mode = batch`.
Sets number of individual events/requests that will be submitted as one HTTP
request by HTTP sink. The default value is 500 which is same as HTTP Sink
`maxBatchSize` |
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java
index 10f9906..7f167e6 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/AbstractRequestSubmitter.java
@@ -19,9 +19,11 @@ package org.apache.flink.connector.http.sink.httpclient;
import org.apache.flink.connector.http.config.HttpConnectorConfigConstants;
import org.apache.flink.connector.http.utils.ThreadUtils;
+import org.apache.flink.util.TimeUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -31,12 +33,12 @@ public abstract class AbstractRequestSubmitter implements
RequestSubmitter {
protected static final int HTTP_CLIENT_PUBLISHING_THREAD_POOL_SIZE = 1;
- protected static final String DEFAULT_REQUEST_TIMEOUT_SECONDS = "30";
+ protected static final Duration DEFAULT_REQUEST_TIMEOUT =
Duration.ofSeconds(30);
/** Thread pool to handle HTTP response from HTTP client. */
protected final ExecutorService publishingThreadPool;
- protected final int httpRequestTimeOutSeconds;
+ protected final Duration httpRequestTimeout;
protected final String[] headersAndValues;
@@ -53,11 +55,14 @@ public abstract class AbstractRequestSubmitter implements
RequestSubmitter {
"http-sink-client-response-worker",
ThreadUtils.LOGGING_EXCEPTION_HANDLER));
- this.httpRequestTimeOutSeconds =
- Integer.parseInt(
- properties.getProperty(
-
HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS,
- DEFAULT_REQUEST_TIMEOUT_SECONDS));
+ // Sink submitters are wired with Properties today; reading
SINK_REQUEST_TIMEOUT from
+ // ReadableConfig will be handled in a follow-up sink
config-unification PR.
+ String requestTimeout =
+
properties.getProperty(HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS);
+ this.httpRequestTimeout =
+ requestTimeout == null
+ ? DEFAULT_REQUEST_TIMEOUT
+ : TimeUtils.parseDuration(requestTimeout);
this.httpClient = httpClient;
}
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java
index 6733ba2..b573b0c 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitter.java
@@ -31,7 +31,6 @@ import java.net.http.HttpRequest.BodyPublishers;
import java.net.http.HttpRequest.Builder;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -138,7 +137,7 @@ public class BatchRequestSubmitter extends
AbstractRequestSubmitter {
java.net.http.HttpRequest.newBuilder()
.uri(endpointUri)
.version(Version.HTTP_1_1)
-
.timeout(Duration.ofSeconds(httpRequestTimeOutSeconds))
+ .timeout(httpRequestTimeout)
.method(method, publisher);
if (headersAndValues.length != 0) {
diff --git
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java
index aee85c9..3ab5f6e 100644
---
a/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java
+++
b/flink-connector-http/src/main/java/org/apache/flink/connector/http/sink/httpclient/PerRequestSubmitter.java
@@ -27,7 +27,6 @@ import java.net.http.HttpClient.Version;
import java.net.http.HttpRequest.BodyPublishers;
import java.net.http.HttpRequest.Builder;
import java.net.http.HttpResponse;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
@@ -79,7 +78,7 @@ public class PerRequestSubmitter extends
AbstractRequestSubmitter {
java.net.http.HttpRequest.newBuilder()
.uri(endpointUri)
.version(Version.HTTP_1_1)
- .timeout(Duration.ofSeconds(httpRequestTimeOutSeconds))
+ .timeout(httpRequestTimeout)
.method(
requestEntry.method,
BodyPublishers.ofByteArray(requestEntry.element));
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 7d115c6..a0807f1 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
@@ -44,14 +44,12 @@ import static
org.apache.flink.connector.http.config.HttpConnectorConfigConstant
@Slf4j
public abstract class RequestFactoryBase implements HttpRequestFactory {
- public static final String DEFAULT_REQUEST_TIMEOUT_SECONDS = "30";
-
/** Base url used for {@link HttpRequest} for example
"http://localhost:8080";. */
protected final String baseUrl;
protected final LookupQueryCreator lookupQueryCreator;
- protected final int httpRequestTimeOutSeconds;
+ protected final Duration httpRequestTimeout;
/** HTTP headers that should be used for {@link HttpRequest} created by
factory. */
private final String[] headersAndValues;
@@ -80,18 +78,17 @@ public abstract class RequestFactoryBase implements
HttpRequestFactory {
this.headersAndValues =
HttpHeaderUtils.toHeaderAndValueArray(headerMap);
- this.httpRequestTimeOutSeconds =
- Integer.parseInt(
- options.getProperties()
- .getProperty(
-
HttpConnectorConfigConstants.LOOKUP_HTTP_TIMEOUT_SECONDS,
- DEFAULT_REQUEST_TIMEOUT_SECONDS));
-
- String httpVersionFromConfig = null;
- if (options.getReadableConfig() != null) {
- httpVersionFromConfig =
-
options.getReadableConfig().get(HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION);
- }
+ var readableConfig = options.getReadableConfig();
+ this.httpRequestTimeout =
+ readableConfig != null
+ ? readableConfig.get(
+
HttpLookupConnectorOptions.SOURCE_LOOKUP_REQUEST_TIMEOUT)
+ :
HttpLookupConnectorOptions.SOURCE_LOOKUP_REQUEST_TIMEOUT.defaultValue();
+
+ String httpVersionFromConfig =
+ readableConfig != null
+ ?
readableConfig.get(HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION)
+ : null;
if (httpVersionFromConfig == null) {
httpVersion = null;
} else {
@@ -122,9 +119,7 @@ public abstract class RequestFactoryBase implements
HttpRequestFactory {
* @return {@link Builder} for given lookupQuery.
*/
protected Builder setUpRequestMethod(LookupQueryInfo lookupQuery) {
- HttpRequest.Builder builder =
- HttpRequest.newBuilder()
-
.timeout(Duration.ofSeconds(this.httpRequestTimeOutSeconds));
+ HttpRequest.Builder builder =
HttpRequest.newBuilder().timeout(this.httpRequestTimeout);
return httpVersion == null ? builder : builder.version(httpVersion);
}
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java
index 12705ac..76fad07 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/BatchRequestSubmitterTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.http.sink.httpclient;
import org.apache.flink.connector.http.config.HttpConnectorConfigConstants;
import org.apache.flink.connector.http.sink.HttpSinkRequestEntry;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -30,6 +31,7 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
@@ -37,6 +39,7 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -71,6 +74,35 @@ class BatchRequestSubmitterTest {
verify(mockHttpClient,
times(expectedNumberOfBatchRequests)).sendAsync(any(), any());
}
+ /**
+ * The sink shares the same latent defect as FLINK-39364 (#31): {@code
+ * http.sink.request.timeout} was declared as a {@code durationType()}
option but read with
+ * {@code Integer.parseInt(...)}, so a unit-suffixed value like {@code
"45s"} threw {@link
+ * NumberFormatException}. It must now be parsed as a real {@link
Duration}.
+ */
+ @Test
+ public void requestTimeoutWithUnitIsParsedAsDuration() {
+ Properties properties = new Properties();
+
properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE,
"50");
+
properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_TIMEOUT_SECONDS,
"45s");
+
+ BatchRequestSubmitter submitter =
+ new BatchRequestSubmitter(properties, new String[0],
mockHttpClient);
+
+
assertThat(submitter.httpRequestTimeout).isEqualTo(Duration.ofSeconds(45));
+ }
+
+ @Test
+ public void requestTimeoutDefaultsToThirtySeconds() {
+ Properties properties = new Properties();
+
properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE,
"50");
+
+ BatchRequestSubmitter submitter =
+ new BatchRequestSubmitter(properties, new String[0],
mockHttpClient);
+
+
assertThat(submitter.httpRequestTimeout).isEqualTo(Duration.ofSeconds(30));
+ }
+
private static Stream<Arguments> httpRequestMethods() {
return Stream.of(
Arguments.of(List.of("PUT", "PUT", "PUT", "PUT", "POST"), 2),
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 5e0d973..aba0cfd 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
@@ -20,11 +20,13 @@ package org.apache.flink.connector.http.table.lookup;
import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import java.net.URI;
import java.net.http.HttpClient;
+import java.time.Duration;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
@@ -34,6 +36,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.LOOKUP_HTTP_VERSION;
+import static
org.apache.flink.connector.http.table.lookup.HttpLookupConnectorOptions.SOURCE_LOOKUP_REQUEST_TIMEOUT;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link BodyBasedRequestFactory}. */
@@ -76,6 +79,57 @@ public class BodyBasedRequestFactoryTest {
}
}
+ /**
+ * FLINK-39364 (#31) declared {@code http.source.lookup.request.timeout}
as a {@code
+ * durationType()} option, but the runtime kept reading it with {@code
Integer.parseInt(...)}. A
+ * unit-suffixed value such as {@code "30s"} therefore threw {@link
NumberFormatException} when
+ * the request was built. This asserts the value is now parsed as a real
{@link Duration}.
+ */
+ @Test
+ void requestTimeoutWithSecondUnitIsParsedAsDuration() {
+ assertThat(factoryWithRequestTimeout("30s").httpRequestTimeout)
+ .isEqualTo(Duration.ofSeconds(30));
+ }
+
+ @Test
+ void requestTimeoutSupportsMinuteUnit() {
+ assertThat(factoryWithRequestTimeout("1 min").httpRequestTimeout)
+ .isEqualTo(Duration.ofMinutes(1));
+ }
+
+ @Test
+ void requestTimeoutDefaultsToThirtySeconds() {
+ // Empty Configuration: key unset, so readableConfig.get(...) returns
the option default.
+ RequestFactoryBase factory =
+ new BodyBasedRequestFactory("GET", null, null,
lookupConfig(new Configuration()));
+
assertThat(factory.httpRequestTimeout).isEqualTo(Duration.ofSeconds(30));
+ }
+
+ /**
+ * Documents the intended behavioural change versus the old int-seconds
read: a bare number now
+ * follows the standard Flink {@link Duration} convention and is
interpreted as milliseconds.
+ */
+ @Test
+ void bareNumberIsInterpretedAsMillisecondsPerFlinkDurationConvention() {
+ assertThat(factoryWithRequestTimeout("30").httpRequestTimeout)
+ .isEqualTo(Duration.ofMillis(30));
+ }
+
+ private static RequestFactoryBase factoryWithRequestTimeout(String value) {
+ Configuration configuration =
+
Configuration.fromMap(Map.of(SOURCE_LOOKUP_REQUEST_TIMEOUT.key(), value));
+ return new BodyBasedRequestFactory("GET", null, null,
lookupConfig(configuration));
+ }
+
+ private static HttpLookupConfig lookupConfig(Configuration configuration) {
+ return HttpLookupConfig.builder()
+ .lookupMethod("GET")
+ .url("http://service";)
+ .useAsync(false)
+ .readableConfig(configuration)
+ .build();
+ }
+
private static class TestSpec {
Map<String, String> bodyBasedUrlQueryParams;
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 5497f1d..a2036e4 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
@@ -977,7 +977,9 @@ class HttpLookupTableSourceITCaseTest {
+ "'http.source.lookup.query-creator' =
'http-generic-json-url',"
+ "'http.request.body-template' =
'{\"customerId\":\"1\"}',"
+ "'lookup.cache' = 'NONE',"
- + "'http.source.lookup.request.timeout' = '30',"
+ // Dormant fixture: use Duration syntax ('30s');
active coverage is
+ // testHttpLookupJoinWithDurationRequestTimeout.
+ + "'http.source.lookup.request.timeout' = '30s',"
+ "'http.source.lookup.request.thread-pool.size' =
'8',"
+ "'http.source.lookup.response.thread-pool.size' =
'4'"
+ ")";
@@ -1075,6 +1077,46 @@ class HttpLookupTableSourceITCaseTest {
}
}
+ /**
+ * Regression test for FLINK-40072 (defect introduced by FLINK-39364 /
#31): a lookup table
+ * declared with a Duration-typed request timeout such as {@code '30s'}
must work end to end.
+ * Before the runtime was wired to the {@code
SOURCE_LOOKUP_REQUEST_TIMEOUT} ConfigOption, this
+ * DDL passed validation (the {@code http.} namespace is skipped by {@code
validateExcept}) but
+ * the lookup crashed at query time with {@code NumberFormatException}
from {@code
+ * Integer.parseInt("30s")}.
+ */
+ @Test
+ void testHttpLookupJoinWithDurationRequestTimeout() throws Exception {
+ setupServerStub(wireMockServer);
+
+ String lookupTable =
+ "CREATE TABLE Customers ("
+ + "id STRING,"
+ + "id2 STRING,"
+ + "msg STRING,"
+ + "uuid STRING,"
+ + "details ROW<"
+ + "isActive BOOLEAN,"
+ + "nestedDetails ROW<"
+ + "balance STRING"
+ + ">"
+ + ">"
+ + ") WITH ("
+ + "'format' = 'json',"
+ + "'connector' = 'http',"
+ + "'lookup-method' = 'GET',"
+ + "'url' = 'http://localhost:";
+ + serverPort
+ + "/client',"
+ + "'http.source.lookup.header.Content-Type' =
'application/json',"
+ + "'http.source.lookup.request.timeout' = '30s'"
+ + ")";
+
+ SortedSet<Row> rows = testLookupJoin(lookupTable, 4);
+
+ assertEnrichedRows(rows);
+ }
+
private LookupCache getCache() {
Map<String, LookupCacheManager.RefCountedCache> managedCaches =
LookupCacheManager.getInstance().getManagedCaches();
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java
index b6569e4..ae2caab 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/table/sink/HttpDynamicTableSinkFactoryTest.java
@@ -25,7 +25,9 @@ import
org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
/**
* Unfortunately it seems that Flink is lazy with connector instantiation, so
one has to call INSERT
@@ -125,9 +127,14 @@ public class HttpDynamicTableSinkFactoryTest {
"http://localhost/";,
HttpDynamicSinkConnectorOptions.SINK_REQUEST_TIMEOUT.key());
tEnv.executeSql(withRequestTimeout);
- // Should not throw ValidationException for the option itself
- // (a RuntimeException for actual HTTP calls is fine in test)
- assertThatThrownBy(() -> tEnv.executeSql("INSERT INTO httpTimeout
VALUES (1)").await())
-
.isNotInstanceOf(org.apache.flink.table.api.ValidationException.class);
+ // The duration-typed option must be recognized at DDL time; INSERT
may fail at runtime
+ // (e.g. no HTTP server) but must not raise ValidationException.
+ Throwable insertFailure =
+ catchThrowable(() -> tEnv.executeSql("INSERT INTO httpTimeout
VALUES (1)").await());
+ if (insertFailure != null) {
+ assertThat(insertFailure)
+ .as("request.timeout must be a recognized option")
+ .isNotInstanceOf(ValidationException.class);
+ }
}
}
