This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 70e18b8 [FLINK-40278][connector-http] Document and test that HTTP
sink retries resend the whole batch
70e18b8 is described below
commit 70e18b8d2f61b130a9ab09253f38af278228e965
Author: bharathreddygunapati <[email protected]>
AuthorDate: Sat Aug 1 20:32:19 2026 +0530
[FLINK-40278][connector-http] Document and test that HTTP sink retries
resend the whole batch
Generated-by: Cursor (auto mode) for the initial change; Claude Code
(Claude Fable 5.1) for the review follow-ups
---
docs/content.zh/docs/connectors/datastream/http.md | 21 +-
docs/content.zh/docs/connectors/table/http.md | 21 ++
docs/content/docs/connectors/datastream/http.md | 21 +-
docs/content/docs/connectors/table/http.md | 21 ++
.../sink/httpclient/JavaNetSinkHttpClientTest.java | 285 +++++++++++++++++++++
5 files changed, 367 insertions(+), 2 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/http.md
b/docs/content.zh/docs/connectors/datastream/http.md
index ce9e6f5..7281853 100644
--- a/docs/content.zh/docs/connectors/datastream/http.md
+++ b/docs/content.zh/docs/connectors/datastream/http.md
@@ -37,6 +37,7 @@ The HTTP Sink connector supports the Flink DataStream API.
* [Apache HTTP Connector](#apache-http-connector)
* [Working with HTTP sink Flink streaming
API](#working-with-http-sink-flink-streaming-api)
* [Sink Connector options](#sink-connector-options-)
+ * [Retries and handling errors (Sink)](#retries-and-handling-errors-sink)
* [Batch submission mode](#batch-submission-mode)
* [Single submission mode](#single-submission-mode)
* [Http headers](#http-headers)
@@ -82,7 +83,25 @@ These options are specified on the builder using the
setProperty method.
| 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. |
| http.sink.request.batch.size | optional |
Applicable only for `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` |
-
+### Retries and handling errors (Sink)
+Retries are disabled by default: `http.sink.max-retries` defaults to `0`, so
the first failed request fails the job.
+
+When `http.sink.max-retries` is greater than `0`, the sink retries requests
that fail before a response is received (for
+example, connection errors) and responses whose status code is listed in
`http.sink.retry-codes`. The delay between
+retries is set by `http.sink.retry-strategy.type` and the matching fixed-delay
or exponential-delay options. Any other
+response that is not successful or ignored is not retried and fails the job. A
request that still fails with a retryable
+error after the last retry also fails the job.
+
+With the legacy `http.sink.error.code` or `http.sink.error.code.exclude`
properties, no response status code is retried:
+a status code in the error list fails the job and every other status code is
treated as success. Only requests that fail
+before a response is received are retried.
+
+In batch submission mode, each HTTP request carries several records and gets
one response status, which applies to every
+record in that request. A retryable response retries all of those records, and
a fatal response or an exhausted retry
+fails all of them. The sink does not retry individual records inside a batch.
Use an idempotent endpoint if retried
+records must not be applied twice. When one flush is split into several HTTP
requests (because of
+`http.sink.request.batch.size` or different HTTP methods), each response
applies only to the records in its own request.
+Retried records are batched again, so they can be grouped into different HTTP
requests than in the first attempt.
### Request submission
HTTP Sink by default submits events in batch. The submission mode can be
changed using `http.sink.writer.request.mode` property using `single` or
`batch` as property value.
diff --git a/docs/content.zh/docs/connectors/table/http.md
b/docs/content.zh/docs/connectors/table/http.md
index 6c25f63..75ab264 100644
--- a/docs/content.zh/docs/connectors/table/http.md
+++ b/docs/content.zh/docs/connectors/table/http.md
@@ -61,6 +61,7 @@ The HTTP source connector supports [Lookup
Joins](https://nightlies.apache.org/f
* [HTTP Sink](#http-sink)
* [Sink Connector Options](#sink-connector-options)
* [Sink table HTTP status codes](#sink-table-http-status-codes)
+ * [Retries and handling errors (Sink)](#retries-and-handling-errors-sink)
* [Request submission](#request-submission)
* [Batch submission mode](#batch-submission-mode)
* [Single submission mode](#single-submission-mode)
@@ -632,6 +633,26 @@ If either legacy property is set together with
`http.sink.success-codes` or `htt
- `http.sink.retry-codes` (default `500,503,504`) for retryable responses
- `http.sink.ignored-response-codes` for responses that should be treated as
successful without retrying
+### Retries and handling errors (Sink)
+Retries are disabled by default: `http.sink.max-retries` defaults to `0`, so
the first failed request fails the job.
+
+When `http.sink.max-retries` is greater than `0`, the sink retries requests
that fail before a response is received (for
+example, connection errors) and responses whose status code is listed in
`http.sink.retry-codes`. The delay between
+retries is set by `http.sink.retry-strategy.type` and the matching fixed-delay
or exponential-delay options. Any other
+response that is not successful or ignored is not retried and fails the job. A
request that still fails with a retryable
+error after the last retry also fails the job.
+
+With the legacy `http.sink.error.code` or `http.sink.error.code.exclude`
properties, no response status code is retried:
+a status code in the error list fails the job and every other status code is
treated as success. Only requests that fail
+before a response is received are retried.
+
+In batch submission mode, each HTTP request carries several records and gets
one response status, which applies to every
+record in that request. A retryable response retries all of those records, and
a fatal response or an exhausted retry
+fails all of them. The sink does not retry individual records inside a batch.
Use an idempotent endpoint if retried
+records must not be applied twice. When one flush is split into several HTTP
requests (because of
+`http.sink.request.batch.size` or different HTTP methods), each response
applies only to the records in its own request.
+Retried records are batched again, so they can be grouped into different HTTP
requests than in the first attempt.
+
### Request submission
HTTP Sink by default submits events in batch. The submission mode can be
changed using `http.sink.writer.request.mode` property using `single` or
`batch` as property value.
diff --git a/docs/content/docs/connectors/datastream/http.md
b/docs/content/docs/connectors/datastream/http.md
index ce9e6f5..7281853 100644
--- a/docs/content/docs/connectors/datastream/http.md
+++ b/docs/content/docs/connectors/datastream/http.md
@@ -37,6 +37,7 @@ The HTTP Sink connector supports the Flink DataStream API.
* [Apache HTTP Connector](#apache-http-connector)
* [Working with HTTP sink Flink streaming
API](#working-with-http-sink-flink-streaming-api)
* [Sink Connector options](#sink-connector-options-)
+ * [Retries and handling errors (Sink)](#retries-and-handling-errors-sink)
* [Batch submission mode](#batch-submission-mode)
* [Single submission mode](#single-submission-mode)
* [Http headers](#http-headers)
@@ -82,7 +83,25 @@ These options are specified on the builder using the
setProperty method.
| 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. |
| http.sink.request.batch.size | optional |
Applicable only for `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` |
-
+### Retries and handling errors (Sink)
+Retries are disabled by default: `http.sink.max-retries` defaults to `0`, so
the first failed request fails the job.
+
+When `http.sink.max-retries` is greater than `0`, the sink retries requests
that fail before a response is received (for
+example, connection errors) and responses whose status code is listed in
`http.sink.retry-codes`. The delay between
+retries is set by `http.sink.retry-strategy.type` and the matching fixed-delay
or exponential-delay options. Any other
+response that is not successful or ignored is not retried and fails the job. A
request that still fails with a retryable
+error after the last retry also fails the job.
+
+With the legacy `http.sink.error.code` or `http.sink.error.code.exclude`
properties, no response status code is retried:
+a status code in the error list fails the job and every other status code is
treated as success. Only requests that fail
+before a response is received are retried.
+
+In batch submission mode, each HTTP request carries several records and gets
one response status, which applies to every
+record in that request. A retryable response retries all of those records, and
a fatal response or an exhausted retry
+fails all of them. The sink does not retry individual records inside a batch.
Use an idempotent endpoint if retried
+records must not be applied twice. When one flush is split into several HTTP
requests (because of
+`http.sink.request.batch.size` or different HTTP methods), each response
applies only to the records in its own request.
+Retried records are batched again, so they can be grouped into different HTTP
requests than in the first attempt.
### Request submission
HTTP Sink by default submits events in batch. The submission mode can be
changed using `http.sink.writer.request.mode` property using `single` or
`batch` as property value.
diff --git a/docs/content/docs/connectors/table/http.md
b/docs/content/docs/connectors/table/http.md
index 85822ea..5b0a171 100644
--- a/docs/content/docs/connectors/table/http.md
+++ b/docs/content/docs/connectors/table/http.md
@@ -61,6 +61,7 @@ The HTTP source connector supports [Lookup
Joins](https://nightlies.apache.org/f
* [HTTP Sink](#http-sink)
* [Sink Connector Options](#sink-connector-options)
* [Sink table HTTP status codes](#sink-table-http-status-codes)
+ * [Retries and handling errors (Sink)](#retries-and-handling-errors-sink)
* [Request submission](#request-submission)
* [Batch submission mode](#batch-submission-mode)
* [Single submission mode](#single-submission-mode)
@@ -633,6 +634,26 @@ If either legacy property is set together with
`http.sink.success-codes` or `htt
- `http.sink.retry-codes` (default `500,503,504`) for retryable responses
- `http.sink.ignored-response-codes` for responses that should be treated as
successful without retrying
+### Retries and handling errors (Sink)
+Retries are disabled by default: `http.sink.max-retries` defaults to `0`, so
the first failed request fails the job.
+
+When `http.sink.max-retries` is greater than `0`, the sink retries requests
that fail before a response is received (for
+example, connection errors) and responses whose status code is listed in
`http.sink.retry-codes`. The delay between
+retries is set by `http.sink.retry-strategy.type` and the matching fixed-delay
or exponential-delay options. Any other
+response that is not successful or ignored is not retried and fails the job. A
request that still fails with a retryable
+error after the last retry also fails the job.
+
+With the legacy `http.sink.error.code` or `http.sink.error.code.exclude`
properties, no response status code is retried:
+a status code in the error list fails the job and every other status code is
treated as success. Only requests that fail
+before a response is received are retried.
+
+In batch submission mode, each HTTP request carries several records and gets
one response status, which applies to every
+record in that request. A retryable response retries all of those records, and
a fatal response or an exhausted retry
+fails all of them. The sink does not retry individual records inside a batch.
Use an idempotent endpoint if retried
+records must not be applied twice. When one flush is split into several HTTP
requests (because of
+`http.sink.request.batch.size` or different HTTP methods), each response
applies only to the records in its own request.
+Retried records are batched again, so they can be grouped into different HTTP
requests than in the first attempt.
+
### Request submission
HTTP Sink by default submits events in batch. The submission mode can be
changed using `http.sink.writer.request.mode` property using `single` or
`batch` as property value.
diff --git
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java
index fa70916..9fb5eae 100644
---
a/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java
+++
b/flink-connector-http/src/test/java/org/apache/flink/connector/http/sink/httpclient/JavaNetSinkHttpClientTest.java
@@ -28,6 +28,7 @@ import
org.apache.flink.connector.http.table.sink.Slf4jHttpPostRequestCallback;
import org.apache.flink.connector.http.utils.HttpHeaderUtils;
import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -41,9 +42,12 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.net.http.HttpClient;
import java.net.http.HttpResponse;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
@@ -55,9 +59,12 @@ import static
org.apache.flink.connector.http.table.sink.HttpDynamicSinkConnecto
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Test for {@link JavaNetSinkHttpClient }. */
@@ -94,6 +101,14 @@ class JavaNetSinkHttpClientTest {
lenient().when(httpClientBuilder.executor(any())).thenReturn(httpClientBuilder);
}
+ private final List<JavaNetSinkHttpClient> clientsToClose = new
ArrayList<>();
+
+ @AfterEach
+ public void closeClients() {
+ clientsToClose.forEach(JavaNetSinkHttpClient::close);
+ clientsToClose.clear();
+ }
+
private static Stream<Arguments> provideSubmitterFactory() {
return Stream.of(
Arguments.of(new PerRequestRequestSubmitterFactory()),
@@ -378,6 +393,276 @@ class JavaNetSinkHttpClientTest {
assertThat(submitterClosed).isTrue();
}
+ @Test
+ public void perRequestModeRetriesOnlyTheFailedEntry() {
+ HttpSinkRequestEntry successfulEntry = new
HttpSinkRequestEntry("POST", new byte[] {1});
+ HttpSinkRequestEntry retryableEntry = new HttpSinkRequestEntry("POST",
new byte[] {2});
+ HttpClient httpClient = mock(HttpClient.class);
+ doReturn(httpResponse(200), httpResponse(500), httpResponse(200))
+ .when(httpClient)
+ .sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+
+ Properties properties = new Properties();
+ properties.setProperty(SINK_MAX_RETRIES, "1");
+ properties.setProperty(SINK_RETRY_FIXED_DELAY_DELAY, "1ms");
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(properties),
+ recordingSubmitterFactory(httpClient, false,
attempts));
+
+ var response =
+ client.putRequests(List.of(successfulEntry, retryableEntry),
"http://localhost";)
+ .join();
+
+ assertThat(attempts).hasSize(2);
+ assertSameEntries(attempts.get(0), successfulEntry, retryableEntry);
+ assertSameEntries(attempts.get(1), retryableEntry);
+ assertSameEntries(response.getSuccessfulRequests(), successfulEntry,
retryableEntry);
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ @Test
+ public void batchModeRetriesAllOriginalEntriesOfRetryableBatch() {
+ List<HttpSinkRequestEntry> batch = batchEntries();
+ HttpClient httpClient = mock(HttpClient.class);
+ doReturn(httpResponse(500),
httpResponse(200)).when(httpClient).sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(batchProperties("1")),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response = client.putRequests(batch, "http://localhost";).join();
+
+ verify(httpClient, times(2)).sendAsync(any(), any());
+ assertThat(attempts).hasSize(2);
+ assertSameEntries(attempts.get(0), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertSameEntries(attempts.get(1), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertSameEntries(
+ response.getSuccessfulRequests(), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ @Test
+ public void batchModeReturnsAllEntriesAsRetryableWhenRetriesAreExhausted()
{
+ List<HttpSinkRequestEntry> batch = batchEntries();
+ HttpClient httpClient = mock(HttpClient.class);
+ doReturn(httpResponse(500),
httpResponse(500)).when(httpClient).sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(batchProperties("1")),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response = client.putRequests(batch, "http://localhost";).join();
+
+ assertThat(attempts).hasSize(2);
+ assertSameEntries(attempts.get(1), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertThat(response.getSuccessfulRequests()).isEmpty();
+ assertSameEntries(
+ response.getRetriableFailedRequests(), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ @Test
+ public void batchModeFailsAllEntriesOfFatalBatchWithoutRetry() {
+ List<HttpSinkRequestEntry> batch = batchEntries();
+ HttpClient httpClient = mock(HttpClient.class);
+ doReturn(httpResponse(400)).when(httpClient).sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(batchProperties("1")),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response = client.putRequests(batch, "http://localhost";).join();
+
+ verify(httpClient, times(1)).sendAsync(any(), any());
+ assertThat(attempts).hasSize(1);
+ assertThat(response.getSuccessfulRequests()).isEmpty();
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertSameEntries(
+ response.getFatalFailedRequests(), batch.toArray(new
HttpSinkRequestEntry[0]));
+ }
+
+ @Test
+ public void batchModeCompletesAllEntriesOfIgnoredBatch() {
+ List<HttpSinkRequestEntry> batch = batchEntries();
+ HttpClient httpClient = mock(HttpClient.class);
+ doReturn(httpResponse(404)).when(httpClient).sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+
+ Properties properties = batchProperties("1");
+ properties.setProperty(SINK_HTTP_IGNORED_RESPONSE_CODES.key(), "404");
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(properties),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response = client.putRequests(batch, "http://localhost";).join();
+
+ assertThat(attempts).hasSize(1);
+ assertSameEntries(
+ response.getIgnoredRequests(), batch.toArray(new
HttpSinkRequestEntry[0]));
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ @Test
+ public void batchModeRetriesOnlyTheFailedRequestWhenFlushIsSplit() {
+ HttpSinkRequestEntry first = new HttpSinkRequestEntry("POST", new
byte[] {1});
+ HttpSinkRequestEntry second = new HttpSinkRequestEntry("POST", new
byte[] {2});
+ HttpSinkRequestEntry third = new HttpSinkRequestEntry("POST", new
byte[] {3});
+ HttpClient httpClient = mock(HttpClient.class);
+ // first attempt: [first, second] -> 500, [third] -> 200; retry:
[first, second] -> 200
+ doReturn(httpResponse(500), httpResponse(200), httpResponse(200))
+ .when(httpClient)
+ .sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+ List<List<HttpSinkRequestEntry>> httpRequests = new
CopyOnWriteArrayList<>();
+
+ Properties properties = batchProperties("1");
+
properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE,
"2");
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(properties,
recordingCallback(httpRequests)),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response = client.putRequests(List.of(first, second, third),
"http://localhost";).join();
+
+ assertThat(attempts).hasSize(2);
+ assertSameEntries(attempts.get(0), first, second, third);
+ assertSameEntries(attempts.get(1), first, second);
+ assertThat(httpRequests).hasSize(3);
+ assertSameEntries(httpRequests.get(0), first, second);
+ assertSameEntries(httpRequests.get(1), third);
+ assertSameEntries(httpRequests.get(2), first, second);
+ assertSameEntriesInAnyOrder(response.getSuccessfulRequests(), first,
second, third);
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ @Test
+ public void batchModeRegroupsRetriedEntriesIntoNewHttpRequests() {
+ HttpSinkRequestEntry firstPost = new HttpSinkRequestEntry("POST", new
byte[] {1});
+ HttpSinkRequestEntry put = new HttpSinkRequestEntry("PUT", new byte[]
{2});
+ HttpSinkRequestEntry secondPost = new HttpSinkRequestEntry("POST", new
byte[] {3});
+ HttpClient httpClient = mock(HttpClient.class);
+ // first attempt splits on the method change: [firstPost] -> 500,
[put] -> 200,
+ // [secondPost] -> 500; the retry sends both POST entries as one
request -> 200
+ doReturn(httpResponse(500), httpResponse(200), httpResponse(500),
httpResponse(200))
+ .when(httpClient)
+ .sendAsync(any(), any());
+ List<List<HttpSinkRequestEntry>> attempts = new
CopyOnWriteArrayList<>();
+ List<List<HttpSinkRequestEntry>> httpRequests = new
CopyOnWriteArrayList<>();
+
+ JavaNetSinkHttpClient client =
+ retryingClient(
+ sinkConfig(batchProperties("1"),
recordingCallback(httpRequests)),
+ recordingSubmitterFactory(httpClient, true, attempts));
+
+ var response =
+ client.putRequests(List.of(firstPost, put, secondPost),
"http://localhost";).join();
+
+ verify(httpClient, times(4)).sendAsync(any(), any());
+ assertThat(attempts).hasSize(2);
+ assertSameEntries(attempts.get(1), firstPost, secondPost);
+ assertThat(httpRequests).hasSize(4);
+ assertSameEntries(httpRequests.get(0), firstPost);
+ assertSameEntries(httpRequests.get(1), put);
+ assertSameEntries(httpRequests.get(2), secondPost);
+ assertSameEntries(httpRequests.get(3), firstPost, secondPost);
+ assertSameEntriesInAnyOrder(response.getSuccessfulRequests(),
firstPost, put, secondPost);
+ assertThat(response.getRetriableFailedRequests()).isEmpty();
+ assertThat(response.getFatalFailedRequests()).isEmpty();
+ }
+
+ private JavaNetSinkHttpClient retryingClient(
+ HttpSinkConfig sinkConfig, RequestSubmitterFactory
requestSubmitterFactory) {
+ JavaNetSinkHttpClient client =
+ new JavaNetSinkHttpClient(sinkConfig, headerPreprocessor,
requestSubmitterFactory);
+ clientsToClose.add(client);
+ return client;
+ }
+
+ private static HttpPostRequestCallback<HttpRequest> recordingCallback(
+ List<List<HttpSinkRequestEntry>> httpRequests) {
+ return (response, request, endpointUrl, headerMap) ->
+ httpRequests.add(request.getRequestEntries());
+ }
+
+ private static List<HttpSinkRequestEntry> batchEntries() {
+ return List.of(
+ new HttpSinkRequestEntry("POST", new byte[] {1}),
+ new HttpSinkRequestEntry("POST", new byte[] {2}),
+ new HttpSinkRequestEntry("POST", new byte[] {3}));
+ }
+
+ private static Properties batchProperties(String maxRetries) {
+ Properties properties = new Properties();
+
properties.setProperty(HttpConnectorConfigConstants.SINK_HTTP_BATCH_REQUEST_SIZE,
"50");
+ properties.setProperty(SINK_MAX_RETRIES, maxRetries);
+ properties.setProperty(SINK_RETRY_FIXED_DELAY_DELAY, "1ms");
+ return properties;
+ }
+
+ private static RequestSubmitterFactory recordingSubmitterFactory(
+ HttpClient httpClient, boolean batchMode,
List<List<HttpSinkRequestEntry>> attempts) {
+ return (sinkConfig, headersAndValues) -> {
+ RequestSubmitter delegate =
+ batchMode
+ ? new BatchRequestSubmitter(
+ sinkConfig,
+ headersAndValues,
+ httpClient,
+ Executors.newSingleThreadExecutor())
+ : new PerRequestSubmitter(
+ sinkConfig,
+ headersAndValues,
+ httpClient,
+ Executors.newSingleThreadExecutor());
+ return new RequestSubmitter() {
+ @Override
+ public List<CompletableFuture<JavaNetHttpResponseWrapper>>
submit(
+ String endpointUrl, List<HttpSinkRequestEntry>
requestToSubmit) {
+ attempts.add(List.copyOf(requestToSubmit));
+ return delegate.submit(endpointUrl, requestToSubmit);
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+ };
+ };
+ }
+
+ private static CompletableFuture<HttpResponse<String>> httpResponse(int
statusCode) {
+ HttpResponse<String> response = mock(HttpResponse.class);
+ when(response.statusCode()).thenReturn(statusCode);
+ return CompletableFuture.completedFuture(response);
+ }
+
+ private static void assertSameEntries(
+ List<HttpSinkRequestEntry> actual, HttpSinkRequestEntry...
expected) {
+ assertThat(actual)
+ .usingElementComparator((left, right) -> left == right ? 0 : 1)
+ .containsExactly(expected);
+ }
+
+ private static void assertSameEntriesInAnyOrder(
+ List<HttpSinkRequestEntry> actual, HttpSinkRequestEntry...
expected) {
+ assertThat(actual)
+ .usingElementComparator((left, right) -> left == right ? 0 : 1)
+ .containsExactlyInAnyOrder(expected);
+ }
+
private static CompletableFuture<JavaNetHttpResponseWrapper>
responseFuture(
HttpSinkRequestEntry requestEntry, int statusCode) {
HttpResponse<String> response = mock(HttpResponse.class);
