This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 62d6804fea [Feature][Connector-V2] Add binary format support for HTTP
source connector (#10956)
62d6804fea is described below
commit 62d6804fea24d38a857004d6ccb9c39ca07f52c8
Author: cosmosni <[email protected]>
AuthorDate: Sat May 30 13:29:03 2026 +0800
[Feature][Connector-V2] Add binary format support for HTTP source connector
(#10956)
---
docs/en/connectors/source/Http.md | 37 +++-
docs/zh/connectors/source/Http.md | 37 +++-
.../seatunnel/http/client/HttpClientProvider.java | 206 +++++++++++++++++++++
.../seatunnel/http/client/HttpResponse.java | 28 +++
.../seatunnel/http/config/HttpConfig.java | 3 +-
.../seatunnel/http/config/HttpSourceOptions.java | 7 +
.../http/exception/HttpConnectorErrorCode.java | 3 +-
.../http/source/BinaryResponseCollector.java | 64 +++++++
.../seatunnel/http/source/FilenameExtractor.java | 92 +++++++++
.../seatunnel/http/source/HttpSource.java | 149 ++++++++++++---
.../seatunnel/http/source/HttpSourceReader.java | 80 +++++++-
.../http/source/BinaryResponseCollectorTest.java | 141 ++++++++++++++
.../http/source/FilenameExtractorTest.java | 99 ++++++++++
.../seatunnel/e2e/connector/http/HttpIT.java | 4 +
.../src/test/resources/http_binary_to_assert.conf | 55 ++++++
.../src/test/resources/mockserver-config.json | 16 ++
16 files changed, 985 insertions(+), 36 deletions(-)
diff --git a/docs/en/connectors/source/Http.md
b/docs/en/connectors/source/Http.md
index b373bef9a8..1af464b2bb 100644
--- a/docs/en/connectors/source/Http.md
+++ b/docs/en/connectors/source/Http.md
@@ -60,7 +60,8 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| pageing.cursor_field | String | No | - | this
parameter is used to specify the Cursor field name in the request parameter.
|
| pageing.cursor_response_field | String | No | - | This
parameter specifies the field in the response from which the cursor is
retrieved.
|
| content_field | String | No | - | This
parameter can get some json data.If you only need the data in the 'book'
section, configure `content_field = "$.store.book.*"`.
|
-| format | String | No | text | The
format of upstream data, now only support `json` `text`, default `text`.
|
+| format | String | No | text | The
format of upstream data, supports `json` `text` `binary`, default `text`. When
set to `binary`, the response body is treated as raw bytes for downloading
files (PDF, images, ZIP, etc.). |
+| binary_chunk_size | Long | No | 10485760 | Chunk
size in bytes when `format = binary`. Large files are split into multiple rows.
Default 10MB. Only effective in BATCH mode.
|
| method | String | No | get | Http
request method, only supports GET, POST method.
|
| headers | Map | No | - | Http
headers.
|
| params | Map | No | - | Http
params.
|
@@ -191,6 +192,40 @@ connector will generate data as the following:
|----------------------------------------------------------|
| {"code": 200, "data": "get success", "success": true} |
+when you assign format is `binary`, the HTTP response body is treated as raw
bytes for downloading files (PDF, images, ZIP, etc.). The output schema is
fixed as `(data: bytes, relativePath: string, partIndex: long)`. Large files
are automatically split into multiple rows based on `binary_chunk_size`. Only
supports BATCH mode.
+
+Example: Download a file via HTTP and write to LocalFileSink:
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Http {
+ url = "http://example.com/files/report.pdf";
+ method = "GET"
+ format = "binary"
+ binary_chunk_size = 10485760 # 10MB per chunk
+ schema = {
+ fields {
+ data = bytes
+ relativePath = string
+ partIndex = long
+ }
+ }
+ }
+}
+
+sink {
+ LocalFile {
+ path = "/tmp/download"
+ file_format = "binary"
+ }
+}
+```
+
### keep_params_as_form
For compatibility with old versions of http.
When set to true,`<params>` and `<pageing>` will be submitted in the form.
diff --git a/docs/zh/connectors/source/Http.md
b/docs/zh/connectors/source/Http.md
index 39cedef67c..5ba01bca56 100644
--- a/docs/zh/connectors/source/Http.md
+++ b/docs/zh/connectors/source/Http.md
@@ -50,7 +50,8 @@ import ChangeLog from '../changelog/connector-http.md';
| pageing.cursor_field | String | 否 | - |
此参数用于指定请求参数中的游标字段名称。
|
| pageing.cursor_response_field | String | 否 | - |
此参数指定从中检索游标的响应字段。
|
| content_field | String | 否 | - | 此参数可以获取一些
json 数据。如果您只需要 'book' 部分的数据，配置 `content_field = "$.store.book.*"`。
|
-| format | String | 否 | text |
上游数据的格式，目前仅支持 `json` `text`，默认为 `text`。
|
+| format | String | 否 | text | 上游数据的格式，支持
`json` `text` `binary`，默认为 `text`。当设置为 `binary` 时，响应体作为原始字节处理，用于下载文件（PDF、图片、ZIP
等）。 |
+| binary_chunk_size | Long | 否 | 10485760 | 当 `format
= binary` 时的分片大小（字节）。大文件会被拆分为多行。默认 10MB。仅在 BATCH 模式下生效。
|
| method | String | 否 | get | Http
请求方法，仅支持 GET、POST 方法。
|
| headers | Map | 否 | - | Http 头信息。
|
| params | Map | 否 | - | Http 参数。
|
@@ -180,6 +181,40 @@ schema {
|----------------------------------------------------------|
| {"code": 200, "data": "get success", "success": true} |
+当您指定 format 为 `binary` 时，HTTP 响应体作为原始字节处理，用于下载文件（PDF、图片、ZIP 等）。输出 schema 固定为
`(data: bytes, relativePath: string, partIndex: long)`。大文件会根据
`binary_chunk_size` 自动拆分为多行。仅支持 BATCH 模式。
+
+示例：通过 HTTP 下载文件并写入 LocalFileSink：
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Http {
+ url = "http://example.com/files/report.pdf";
+ method = "GET"
+ format = "binary"
+ binary_chunk_size = 10485760 # 每个分片 10MB
+ schema = {
+ fields {
+ data = bytes
+ relativePath = string
+ partIndex = long
+ }
+ }
+ }
+}
+
+sink {
+ LocalFile {
+ path = "/tmp/download"
+ file_format = "binary"
+ }
+}
+```
+
### keep_params_as_form
为了兼容旧版本的 http。
当设置为 true 时，`<params>` 和 `<pageing>` 将以表单形式提交。
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpClientProvider.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpClientProvider.java
index 5b4075dea5..964a402013 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpClientProvider.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpClientProvider.java
@@ -169,6 +169,191 @@ public class HttpClientProvider implements AutoCloseable {
return doGet(url, headers, params);
}
+ public HttpResponse executeBinary(
+ String url,
+ String method,
+ Map<String, String> headers,
+ Map<String, String> params,
+ String body,
+ boolean keepParamsAsForm)
+ throws Exception {
+ method = method.toUpperCase(Locale.ROOT);
+ if (HttpGet.METHOD_NAME.equals(method)) {
+ URIBuilder uriBuilder = new URIBuilder(url);
+ addParameters(uriBuilder, params);
+ HttpGet httpGet = new HttpGet(uriBuilder.build());
+ httpGet.setConfig(requestConfig);
+ addHeaders(httpGet, headers);
+ return getResponseBinary(httpGet);
+ }
+ if (HttpPost.METHOD_NAME.equals(method)) {
+ if (keepParamsAsForm) {
+ return doPostBinary(url, headers, params, body);
+ }
+ HttpPost httpPost = new HttpPost(url);
+ httpPost.setConfig(requestConfig);
+ addHeaders(httpPost, headers);
+ if (!Strings.isNullOrEmpty(body)) {
+ httpPost.setEntity(new StringEntity(body,
ContentType.APPLICATION_JSON));
+ } else {
+ addParameters(httpPost, params);
+ }
+ return getResponseBinary(httpPost);
+ }
+ // fallback to GET for other methods
+ URIBuilder uriBuilder = new URIBuilder(url);
+ addParameters(uriBuilder, params);
+ HttpGet httpGet = new HttpGet(uriBuilder.build());
+ httpGet.setConfig(requestConfig);
+ addHeaders(httpGet, headers);
+ return getResponseBinary(httpGet);
+ }
+
+ private HttpResponse doPostBinary(
+ String url, Map<String, String> headers, Map<String, String>
params, String body)
+ throws Exception {
+ Map<String, Object> bodyMap = new HashMap<>();
+ if (!Strings.isNullOrEmpty(body)) {
+ bodyMap =
+ ConfigFactory.parseString(body).entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
entry.getValue().unwrapped(),
+ (v1, v2) -> v2));
+ }
+ if (MapUtils.isNotEmpty(params)) {
+ headers = MapUtils.isEmpty(headers) ? new HashMap<>() : headers;
+ headers.putIfAbsent(HTTP.CONTENT_TYPE, APPLICATION_FORM);
+ }
+ if (MapUtils.isEmpty(bodyMap)) {
+ bodyMap = new HashMap<>();
+ }
+ bodyMap.putAll(params);
+ URIBuilder uriBuilder = new URIBuilder(url);
+ HttpPost httpPost = new HttpPost(uriBuilder.build());
+ httpPost.setConfig(requestConfig);
+ addHeaders(httpPost, headers);
+ addBody(httpPost, bodyMap);
+ return getResponseBinary(httpPost);
+ }
+
+ /**
+ * Streams binary response directly in chunks without buffering the entire
body. The consumer is
+ * called once per chunk with a SeaTunnelRow containing (byte[] data,
String filename, long
+ * partIndex).
+ */
+ public int executeBinaryStreaming(
+ String url,
+ String method,
+ Map<String, String> headers,
+ Map<String, String> params,
+ String body,
+ boolean keepParamsAsForm,
+ long chunkSize,
+ String urlForFilename,
+ java.util.function.Consumer<Object[]> chunkConsumer)
+ throws Exception {
+ HttpRequestBase request =
+ buildRequest(url, method, headers, params, body,
keepParamsAsForm);
+ try (CloseableHttpResponse response = retryWithException(request)) {
+ if (response == null
+ || response.getStatusLine() == null
+ || response.getStatusLine().getStatusCode() >= 300) {
+ int code =
+ response != null && response.getStatusLine() != null
+ ? response.getStatusLine().getStatusCode()
+ : HttpStatus.SC_INTERNAL_SERVER_ERROR;
+ return code;
+ }
+
+ if (response.getEntity() == null) {
+ return response.getStatusLine().getStatusCode();
+ }
+
+ String contentDisposition = null;
+ if (response.getFirstHeader("Content-Disposition") != null) {
+ contentDisposition =
response.getFirstHeader("Content-Disposition").getValue();
+ }
+ String filename =
+
org.apache.seatunnel.connectors.seatunnel.http.source.FilenameExtractor.extract(
+ contentDisposition, urlForFilename);
+
+ try (java.io.InputStream in = response.getEntity().getContent()) {
+ byte[] buffer = new byte[(int) chunkSize];
+ long partIndex = 0;
+ int bytesRead;
+ while ((bytesRead = in.read(buffer)) > 0) {
+ byte[] chunk = new byte[bytesRead];
+ System.arraycopy(buffer, 0, chunk, 0, bytesRead);
+ chunkConsumer.accept(new Object[] {chunk, filename,
partIndex});
+ partIndex++;
+ }
+ }
+ return response.getStatusLine().getStatusCode();
+ }
+ }
+
+ private HttpRequestBase buildRequest(
+ String url,
+ String method,
+ Map<String, String> headers,
+ Map<String, String> params,
+ String body,
+ boolean keepParamsAsForm)
+ throws Exception {
+ method = method.toUpperCase(Locale.ROOT);
+ if (HttpGet.METHOD_NAME.equals(method)) {
+ URIBuilder uriBuilder = new URIBuilder(url);
+ addParameters(uriBuilder, params);
+ HttpGet httpGet = new HttpGet(uriBuilder.build());
+ httpGet.setConfig(requestConfig);
+ addHeaders(httpGet, headers);
+ return httpGet;
+ }
+ if (HttpPost.METHOD_NAME.equals(method)) {
+ HttpPost httpPost = new HttpPost(url);
+ httpPost.setConfig(requestConfig);
+ if (keepParamsAsForm) {
+ Map<String, Object> bodyMap = new HashMap<>();
+ if (!Strings.isNullOrEmpty(body)) {
+ bodyMap =
+ ConfigFactory.parseString(body).entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
entry.getValue().unwrapped(),
+ (v1, v2) -> v2));
+ }
+ if (MapUtils.isNotEmpty(params)) {
+ headers = MapUtils.isEmpty(headers) ? new HashMap<>() :
headers;
+ headers.putIfAbsent(HTTP.CONTENT_TYPE, APPLICATION_FORM);
+ }
+ if (MapUtils.isEmpty(bodyMap)) {
+ bodyMap = new HashMap<>();
+ }
+ bodyMap.putAll(params);
+ addHeaders(httpPost, headers);
+ addBody(httpPost, bodyMap);
+ } else {
+ addHeaders(httpPost, headers);
+ if (!Strings.isNullOrEmpty(body)) {
+ httpPost.setEntity(new StringEntity(body,
ContentType.APPLICATION_JSON));
+ } else {
+ addParameters(httpPost, params);
+ }
+ }
+ return httpPost;
+ }
+ // fallback to GET for other methods
+ URIBuilder uriBuilder = new URIBuilder(url);
+ addParameters(uriBuilder, params);
+ HttpGet httpGet = new HttpGet(uriBuilder.build());
+ httpGet.setConfig(requestConfig);
+ addHeaders(httpGet, headers);
+ return httpGet;
+ }
+
/**
* Send a get request without request headers and request parameters
*
@@ -430,6 +615,27 @@ public class HttpClientProvider implements AutoCloseable {
return new HttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR);
}
+ private HttpResponse getResponseBinary(HttpRequestBase request) throws
Exception {
+ try (CloseableHttpResponse httpResponse = retryWithException(request))
{
+ if (httpResponse != null && httpResponse.getStatusLine() != null) {
+ byte[] bodyBytes = null;
+ if (httpResponse.getEntity() != null) {
+ bodyBytes =
EntityUtils.toByteArray(httpResponse.getEntity());
+ }
+ String contentDisposition = null;
+ if (httpResponse.getFirstHeader("Content-Disposition") !=
null) {
+ contentDisposition =
+
httpResponse.getFirstHeader("Content-Disposition").getValue();
+ }
+ return new HttpResponse(
+ httpResponse.getStatusLine().getStatusCode(),
+ bodyBytes,
+ contentDisposition);
+ }
+ }
+ return new HttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+ }
+
private CloseableHttpResponse retryWithException(HttpRequestBase request)
throws Exception {
return retryer.call(() -> httpClient.execute(request));
}
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpResponse.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpResponse.java
index cd9d0471ed..cebbb33aeb 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpResponse.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/client/HttpResponse.java
@@ -32,6 +32,12 @@ public class HttpResponse implements Serializable {
/** response body */
private String content;
+ /** binary response body */
+ private byte[] bodyBytes;
+
+ /** Content-Disposition header value */
+ private String contentDisposition;
+
public HttpResponse() {}
public HttpResponse(int code) {
@@ -47,6 +53,12 @@ public class HttpResponse implements Serializable {
this.content = content;
}
+ public HttpResponse(int code, byte[] bodyBytes, String contentDisposition)
{
+ this.code = code;
+ this.bodyBytes = bodyBytes;
+ this.contentDisposition = contentDisposition;
+ }
+
public int getCode() {
return code;
}
@@ -63,6 +75,22 @@ public class HttpResponse implements Serializable {
this.content = content;
}
+ public byte[] getBodyBytes() {
+ return bodyBytes;
+ }
+
+ public void setBodyBytes(byte[] bodyBytes) {
+ this.bodyBytes = bodyBytes;
+ }
+
+ public String getContentDisposition() {
+ return contentDisposition;
+ }
+
+ public void setContentDisposition(String contentDisposition) {
+ this.contentDisposition = contentDisposition;
+ }
+
@Override
public String toString() {
return "HttpClientResult [code=" + code + ", content=" + content + "]";
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java
index 1801dfd0a7..964e14eea4 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpConfig.java
@@ -23,7 +23,8 @@ public class HttpConfig {
public enum ResponseFormat {
JSON("json"),
- TEXT("text");
+ TEXT("text"),
+ BINARY("binary");
private String format;
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSourceOptions.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSourceOptions.java
index 5014a01c4a..5efce88e09 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSourceOptions.java
@@ -110,6 +110,13 @@ public class HttpSourceOptions extends HttpCommonOptions {
.enumType(HttpConfig.ResponseFormat.class)
.defaultValue(HttpConfig.ResponseFormat.TEXT)
.withDescription("Http response format");
+
+ public static final Option<Long> BINARY_CHUNK_SIZE =
+ Options.key("binary_chunk_size")
+ .longType()
+ .defaultValue(10 * 1024 * 1024L)
+ .withDescription(
+ "Chunk size in bytes for binary response mode.
When the response body exceeds this size, it will be split into multiple rows.
Only effective when format=binary. Default is 10MB.");
public static final Option<Integer> POLL_INTERVAL_MILLS =
Options.key("poll_interval_millis")
.intType()
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/exception/HttpConnectorErrorCode.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/exception/HttpConnectorErrorCode.java
index bbea694ce9..83254cb3d4 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/exception/HttpConnectorErrorCode.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/exception/HttpConnectorErrorCode.java
@@ -21,7 +21,8 @@ import
org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
public enum HttpConnectorErrorCode implements SeaTunnelErrorCode {
FIELD_DATA_IS_INCONSISTENT("HTTP-01", "The field data is inconsistent"),
- REQUEST_FAILED("HTTP-02", "The request is failed");
+ REQUEST_FAILED("HTTP-02", "The request is failed"),
+ CONFIG_VALIDATION_FAILED("HTTP-03", "The config validation is failed");
private final String code;
private final String description;
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollector.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollector.java
new file mode 100644
index 0000000000..8fa84e3334
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollector.java
@@ -0,0 +1,64 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.http.source;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse;
+
+public class BinaryResponseCollector {
+
+ public static final SeaTunnelRowType BINARY_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"data", "relativePath", "partIndex"},
+ new SeaTunnelDataType[] {
+ PrimitiveByteArrayType.INSTANCE,
BasicType.STRING_TYPE, BasicType.LONG_TYPE
+ });
+
+ private BinaryResponseCollector() {}
+
+ public static void collect(
+ HttpResponse response, String url, long chunkSize,
Collector<SeaTunnelRow> output) {
+ String filename =
FilenameExtractor.extract(response.getContentDisposition(), url);
+
+ byte[] bodyBytes = response.getBodyBytes();
+ if (bodyBytes == null || bodyBytes.length == 0) {
+ return;
+ }
+
+ if (chunkSize <= 0 || bodyBytes.length <= chunkSize) {
+ output.collect(new SeaTunnelRow(new Object[] {bodyBytes, filename,
0L}));
+ return;
+ }
+
+ int offset = 0;
+ long partIndex = 0;
+ while (offset < bodyBytes.length) {
+ int end = Math.min(offset + (int) chunkSize, bodyBytes.length);
+ byte[] chunk = new byte[end - offset];
+ System.arraycopy(bodyBytes, offset, chunk, 0, end - offset);
+ output.collect(new SeaTunnelRow(new Object[] {chunk, filename,
partIndex}));
+ offset = end;
+ partIndex++;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractor.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractor.java
new file mode 100644
index 0000000000..8511172e4b
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractor.java
@@ -0,0 +1,92 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.http.source;
+
+import java.net.URI;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class FilenameExtractor {
+
+ private static final Pattern FILENAME_PATTERN =
+ Pattern.compile("filename\\*?=([^;]+)", Pattern.CASE_INSENSITIVE);
+ private static final Pattern ENCODED_FILENAME_PATTERN =
+ Pattern.compile("filename\\*=UTF-8''(.+?)(?:;|$)",
Pattern.CASE_INSENSITIVE);
+
+ private FilenameExtractor() {}
+
+ public static String extract(String contentDisposition, String url) {
+ // 1. Try Content-Disposition header
+ if (contentDisposition != null && !contentDisposition.isEmpty()) {
+ String filename = parseContentDisposition(contentDisposition);
+ if (filename != null && !filename.isEmpty()) {
+ return filename;
+ }
+ }
+
+ // 2. Try URL path
+ if (url != null && !url.isEmpty()) {
+ String filename = extractFromUrl(url);
+ if (filename != null && !filename.isEmpty()) {
+ return filename;
+ }
+ }
+
+ // 3. Default
+ return "response-body-" + System.currentTimeMillis();
+ }
+
+ static String parseContentDisposition(String contentDisposition) {
+ // Try RFC 5987 encoded filename first (filename*=UTF-8''xxx)
+ Matcher encodedMatcher =
ENCODED_FILENAME_PATTERN.matcher(contentDisposition);
+ if (encodedMatcher.find()) {
+ try {
+ return java.net.URLDecoder.decode(encodedMatcher.group(1),
"UTF-8");
+ } catch (java.io.UnsupportedEncodingException e) {
+ return encodedMatcher.group(1);
+ }
+ }
+
+ // Try standard filename="xxx"
+ Matcher matcher = FILENAME_PATTERN.matcher(contentDisposition);
+ if (matcher.find()) {
+ String filename = matcher.group(1).trim();
+ if (filename.startsWith("\"") && filename.endsWith("\"")) {
+ filename = filename.substring(1, filename.length() - 1);
+ }
+ return filename;
+ }
+ return null;
+ }
+
+ static String extractFromUrl(String url) {
+ try {
+ String path = URI.create(url).getPath();
+ if (path != null && !path.isEmpty()) {
+ int lastSlash = path.lastIndexOf('/');
+ String filename = lastSlash >= 0 ? path.substring(lastSlash +
1) : path;
+ if (!filename.isEmpty() && filename.contains(".")) {
+ return filename;
+ }
+ }
+ } catch (IllegalArgumentException e) {
+ // URL parsing failed, ignore
+ }
+ return null;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java
index 44629f4f75..7d577b8dfe 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSource.java
@@ -33,7 +33,9 @@ import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.constants.JobMode;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.common.utils.JsonUtils;
@@ -58,6 +60,8 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
protected String contentField;
protected JobContext jobContext;
protected DeserializationSchema<SeaTunnelRow> deserializationSchema;
+ protected boolean binaryMode = false;
+ protected long binaryChunkSize =
HttpSourceOptions.BINARY_CHUNK_SIZE.defaultValue();
protected CatalogTable catalogTable;
@@ -72,13 +76,6 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
return HttpConfig.CONNECTOR_IDENTITY;
}
- @Override
- public Boundedness getBoundedness() {
- return JobMode.BATCH.equals(jobContext.getJobMode())
- ? Boundedness.BOUNDED
- : Boundedness.UNBOUNDED;
- }
-
private void buildPagingWithConfig(ReadonlyConfig config) {
Config pluginConfig = config.toConfig();
if (pluginConfig.hasPath(HttpSourceOptions.PAGEING.key())) {
@@ -145,6 +142,46 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
contentField =
config.getString(HttpSourceOptions.CONTENT_FIELD.key());
}
break;
+ case BINARY:
+ this.binaryMode = true;
+ this.binaryChunkSize =
pluginConfig.get(HttpSourceOptions.BINARY_CHUNK_SIZE);
+ SeaTunnelRowType binaryRowType =
BinaryResponseCollector.BINARY_ROW_TYPE;
+ TableSchema binarySchema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "data",
+
PrimitiveByteArrayType.INSTANCE,
+ 0,
+ false,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "relativePath",
+ BasicType.STRING_TYPE,
+ 0,
+ false,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "partIndex",
+ BasicType.LONG_TYPE,
+ 0,
+ false,
+ null,
+ null))
+ .build();
+ this.catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of(
+ HttpConfig.CONNECTOR_IDENTITY,
TablePath.DEFAULT),
+ binarySchema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null);
+ break;
default:
// TODO: use format SPI
throw new HttpConnectorException(
@@ -154,24 +191,70 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
format));
}
} else {
- TableIdentifier tableIdentifier =
- TableIdentifier.of(HttpConfig.CONNECTOR_IDENTITY,
TablePath.DEFAULT);
- TableSchema tableSchema =
- TableSchema.builder()
- .column(
- PhysicalColumn.of(
- "content", BasicType.STRING_TYPE,
0, false, null, null))
- .build();
-
- this.catalogTable =
- CatalogTable.of(
- tableIdentifier,
- tableSchema,
- Collections.emptyMap(),
- Collections.emptyList(),
- null);
- this.deserializationSchema =
- new
SimpleTextDeserializationSchema(catalogTable.getSeaTunnelRowType());
+ HttpConfig.ResponseFormat format =
pluginConfig.get(HttpSourceOptions.FORMAT);
+ if (format == HttpConfig.ResponseFormat.BINARY) {
+ this.binaryMode = true;
+ this.binaryChunkSize =
pluginConfig.get(HttpSourceOptions.BINARY_CHUNK_SIZE);
+ TableSchema binarySchema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "data",
+
PrimitiveByteArrayType.INSTANCE,
+ 0,
+ false,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "relativePath",
+ BasicType.STRING_TYPE,
+ 0,
+ false,
+ null,
+ null))
+ .column(
+ PhysicalColumn.of(
+ "partIndex",
+ BasicType.LONG_TYPE,
+ 0,
+ false,
+ null,
+ null))
+ .build();
+ this.catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of(
+ HttpConfig.CONNECTOR_IDENTITY,
TablePath.DEFAULT),
+ binarySchema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null);
+ } else {
+ TableIdentifier tableIdentifier =
+ TableIdentifier.of(HttpConfig.CONNECTOR_IDENTITY,
TablePath.DEFAULT);
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "content",
+ BasicType.STRING_TYPE,
+ 0,
+ false,
+ null,
+ null))
+ .build();
+
+ this.catalogTable =
+ CatalogTable.of(
+ tableIdentifier,
+ tableSchema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null);
+ this.deserializationSchema =
+ new
SimpleTextDeserializationSchema(catalogTable.getSeaTunnelRowType());
+ }
}
}
@@ -180,6 +263,18 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
this.jobContext = jobContext;
}
+ @Override
+ public Boundedness getBoundedness() {
+ if (binaryMode && JobMode.STREAMING.equals(jobContext.getJobMode())) {
+ throw new HttpConnectorException(
+ CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
+ "HTTP binary format only supports BATCH mode, not
STREAMING.");
+ }
+ return JobMode.BATCH.equals(jobContext.getJobMode())
+ ? Boundedness.BOUNDED
+ : Boundedness.UNBOUNDED;
+ }
+
@Override
public List<CatalogTable> getProducedCatalogTables() {
return Lists.newArrayList(catalogTable);
@@ -194,7 +289,9 @@ public class HttpSource extends
AbstractSingleSplitSource<SeaTunnelRow> {
this.deserializationSchema,
jsonField,
contentField,
- pageInfo);
+ pageInfo,
+ binaryMode,
+ binaryChunkSize);
}
protected JsonField getJsonField(Config jsonFieldConf) {
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
index c7ab3c9c00..2bc66a3796 100644
---
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceReader.java
@@ -75,6 +75,8 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
Configuration.defaultConfiguration().addOptions(DEFAULT_OPTIONS);
private boolean noMoreElementFlag = true;
private Optional<PageInfo> pageInfoOptional = Optional.empty();
+ private final boolean binaryMode;
+ private final long binaryChunkSize;
/**
* Holds the original request body template for placeholder replacement.
This ensures that the
* state is not unintentionally mutated during pagination.
@@ -87,12 +89,15 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
DeserializationSchema<SeaTunnelRow> deserializationSchema,
JsonField jsonField,
String contentJson) {
- this.context = context;
- this.httpParameter = httpParameter;
- this.deserializationCollector = new
DeserializationCollector(deserializationSchema);
- this.jsonField = jsonField;
- this.contentJson = contentJson;
- this.rawBody = httpParameter.getBody();
+ this(
+ httpParameter,
+ context,
+ deserializationSchema,
+ jsonField,
+ contentJson,
+ null,
+ false,
+ 0L);
}
public HttpSourceReader(
@@ -102,6 +107,26 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
JsonField jsonField,
String contentJson,
PageInfo pageInfo) {
+ this(
+ httpParameter,
+ context,
+ deserializationSchema,
+ jsonField,
+ contentJson,
+ pageInfo,
+ false,
+ 0L);
+ }
+
+ public HttpSourceReader(
+ HttpParameter httpParameter,
+ SingleSplitReaderContext context,
+ DeserializationSchema<SeaTunnelRow> deserializationSchema,
+ JsonField jsonField,
+ String contentJson,
+ PageInfo pageInfo,
+ boolean binaryMode,
+ long binaryChunkSize) {
this.context = context;
this.httpParameter = httpParameter;
this.deserializationCollector = new
DeserializationCollector(deserializationSchema);
@@ -109,6 +134,15 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
this.contentJson = contentJson;
this.pageInfoOptional = Optional.ofNullable(pageInfo);
this.rawBody = httpParameter.getBody();
+ this.binaryMode = binaryMode;
+ this.binaryChunkSize = binaryChunkSize;
+
+ if (binaryMode && pageInfo != null) {
+ throw new HttpConnectorException(
+ HttpConnectorErrorCode.CONFIG_VALIDATION_FAILED,
+ "Binary mode does not support pagination. "
+ + "Please remove pagination configuration or use
text mode.");
+ }
}
@Override
@@ -124,6 +158,10 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
}
public void pollAndCollectData(Collector<SeaTunnelRow> output) throws
Exception {
+ if (binaryMode) {
+ pollAndCollectBinaryData(output);
+ return;
+ }
HttpResponse response = executeRequest();
if (response.getCode() >= 200 && response.getCode() <= 207) {
String content = response.getContent();
@@ -153,6 +191,36 @@ public class HttpSourceReader extends
AbstractSingleSplitReader<SeaTunnelRow> {
}
}
+ private void pollAndCollectBinaryData(Collector<SeaTunnelRow> output)
throws Exception {
+ int statusCode =
+ httpClient.executeBinaryStreaming(
+ this.httpParameter.getUrl(),
+ this.httpParameter.getMethod().getMethod(),
+ this.httpParameter.getHeaders(),
+ this.httpParameter.getParams(),
+ this.httpParameter.getBody(),
+ this.httpParameter.isKeepParamsAsForm(),
+ binaryChunkSize,
+ httpParameter.getUrl(),
+ row -> output.collect(new SeaTunnelRow(row)));
+ if (statusCode >= 200 && statusCode <= 207) {
+ noMoreElementFlag = true;
+ } else {
+ String msg = String.format("http binary request failed, status
code:[%s]", statusCode);
+ throw new
HttpConnectorException(HttpConnectorErrorCode.REQUEST_FAILED, msg);
+ }
+ }
+
+ protected HttpResponse executeBinaryRequest() throws Exception {
+ return httpClient.executeBinary(
+ this.httpParameter.getUrl(),
+ this.httpParameter.getMethod().getMethod(),
+ this.httpParameter.getHeaders(),
+ this.httpParameter.getParams(),
+ this.httpParameter.getBody(),
+ this.httpParameter.isKeepParamsAsForm());
+ }
+
protected HttpResponse executeRequest() throws Exception {
return httpClient.execute(
this.httpParameter.getUrl(),
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollectorTest.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollectorTest.java
new file mode 100644
index 0000000000..747884729b
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/BinaryResponseCollectorTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.http.source;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class BinaryResponseCollectorTest {
+
+ private static Collector<SeaTunnelRow> listCollector(List<SeaTunnelRow>
list) {
+ return new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow record) {
+ list.add(record);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ };
+ }
+
+ @Test
+ public void testSingleRowOutput() {
+ byte[] data = new byte[1024];
+ HttpResponse response = new HttpResponse(200, data, "attachment;
filename=\"test.pdf\"");
+ List<SeaTunnelRow> collected = new ArrayList<>();
+
+ BinaryResponseCollector.collect(
+ response, "http://example.com/file";, 10 * 1024 * 1024L,
listCollector(collected));
+
+ Assertions.assertEquals(1, collected.size());
+ SeaTunnelRow row = collected.get(0);
+ Assertions.assertArrayEquals(data, (byte[]) row.getField(0));
+ Assertions.assertEquals("test.pdf", row.getField(1));
+ Assertions.assertEquals(0L, row.getField(2));
+ }
+
+ @Test
+ public void testChunkedOutput() {
+ byte[] data = new byte[25 * 1024 * 1024]; // 25MB
+ for (int i = 0; i < data.length; i++) {
+ data[i] = (byte) (i % 256);
+ }
+ HttpResponse response = new HttpResponse(200, data, null);
+ List<SeaTunnelRow> collected = new ArrayList<>();
+
+ BinaryResponseCollector.collect(
+ response,
+ "http://example.com/files/big.zip";,
+ 10 * 1024 * 1024L,
+ listCollector(collected));
+
+ Assertions.assertEquals(3, collected.size());
+
+ // First chunk: 10MB
+ SeaTunnelRow chunk0 = collected.get(0);
+ Assertions.assertEquals(10 * 1024 * 1024, ((byte[])
chunk0.getField(0)).length);
+ Assertions.assertEquals(0L, chunk0.getField(2));
+
+ // Second chunk: 10MB
+ SeaTunnelRow chunk1 = collected.get(1);
+ Assertions.assertEquals(10 * 1024 * 1024, ((byte[])
chunk1.getField(0)).length);
+ Assertions.assertEquals(1L, chunk1.getField(2));
+
+ // Third chunk: 5MB
+ SeaTunnelRow chunk2 = collected.get(2);
+ Assertions.assertEquals(5 * 1024 * 1024, ((byte[])
chunk2.getField(0)).length);
+ Assertions.assertEquals(2L, chunk2.getField(2));
+
+ // All chunks have same filename
+ String filename = (String) chunk0.getField(1);
+ Assertions.assertEquals("big.zip", filename);
+ Assertions.assertEquals(filename, chunk1.getField(1));
+ Assertions.assertEquals(filename, chunk2.getField(1));
+
+ // Verify data integrity: reassemble and compare
+ byte[] reassembled = new byte[data.length];
+ for (SeaTunnelRow row : collected) {
+ byte[] chunk = (byte[]) row.getField(0);
+ int partIndex = ((Long) row.getField(2)).intValue();
+ System.arraycopy(chunk, 0, reassembled, partIndex * 10 * 1024 *
1024, chunk.length);
+ }
+ Assertions.assertArrayEquals(data, reassembled);
+ }
+
+ @Test
+ public void testNullBodyBytes() {
+ HttpResponse response = new HttpResponse(200, null, null);
+ List<SeaTunnelRow> collected = new ArrayList<>();
+
+ BinaryResponseCollector.collect(
+ response, "http://example.com/file";, 1024L,
listCollector(collected));
+
+ Assertions.assertEquals(0, collected.size());
+ }
+
+ @Test
+ public void testEmptyBodyBytes() {
+ HttpResponse response = new HttpResponse(200, new byte[0], null);
+ List<SeaTunnelRow> collected = new ArrayList<>();
+
+ BinaryResponseCollector.collect(
+ response, "http://example.com/file";, 1024L,
listCollector(collected));
+
+ Assertions.assertEquals(0, collected.size());
+ }
+
+ @Test
+ public void testBinaryRowTypeStructure() {
+ Assertions.assertEquals(3,
BinaryResponseCollector.BINARY_ROW_TYPE.getTotalFields());
+ Assertions.assertEquals("data",
BinaryResponseCollector.BINARY_ROW_TYPE.getFieldName(0));
+ Assertions.assertEquals(
+ "relativePath",
BinaryResponseCollector.BINARY_ROW_TYPE.getFieldName(1));
+ Assertions.assertEquals(
+ "partIndex",
BinaryResponseCollector.BINARY_ROW_TYPE.getFieldName(2));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractorTest.java
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractorTest.java
new file mode 100644
index 0000000000..34d36609ca
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/source/FilenameExtractorTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.http.source;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FilenameExtractorTest {
+
+ @Test
+ public void testParseContentDispositionStandard() {
+ String result =
+ FilenameExtractor.parseContentDisposition("attachment;
filename=\"report.pdf\"");
+ Assertions.assertEquals("report.pdf", result);
+ }
+
+ @Test
+ public void testParseContentDispositionWithoutQuotes() {
+ String result =
+ FilenameExtractor.parseContentDisposition("attachment;
filename=report.pdf");
+ Assertions.assertEquals("report.pdf", result);
+ }
+
+ @Test
+ public void testParseContentDispositionUtf8Encoded() {
+ String result =
+ FilenameExtractor.parseContentDisposition(
+ "attachment; filename*=UTF-8''%E6%8A%A5%E5%91%8A.pdf");
+ Assertions.assertEquals("报告.pdf", result);
+ }
+
+ @Test
+ public void testParseContentDispositionNoFilename() {
+ String result =
FilenameExtractor.parseContentDisposition("attachment");
+ Assertions.assertNull(result);
+ }
+
+ @Test
+ public void testExtractFromUrlNormalPath() {
+ String result =
FilenameExtractor.extractFromUrl("http://example.com/files/demo.pdf";);
+ Assertions.assertEquals("demo.pdf", result);
+ }
+
+ @Test
+ public void testExtractFromUrlNoExtension() {
+ String result =
FilenameExtractor.extractFromUrl("http://example.com/api/download";);
+ Assertions.assertNull(result);
+ }
+
+ @Test
+ public void testExtractFromUrlWithQueryParams() {
+ String result =
+ FilenameExtractor.extractFromUrl(
+ "http://example.com/files/archive.zip?token=abc123";);
+ Assertions.assertEquals("archive.zip", result);
+ }
+
+ @Test
+ public void testExtractFromUrlTrailingSlash() {
+ String result =
FilenameExtractor.extractFromUrl("http://example.com/files/";);
+ Assertions.assertNull(result);
+ }
+
+ @Test
+ public void testExtractPriorityContentDispositionOverUrl() {
+ String result =
+ FilenameExtractor.extract(
+ "attachment; filename=\"report.pdf\"",
+ "http://example.com/files/other-name.pdf";);
+ Assertions.assertEquals("report.pdf", result);
+ }
+
+ @Test
+ public void testExtractFallbackToUrl() {
+ String result = FilenameExtractor.extract(null,
"http://example.com/files/demo.zip";);
+ Assertions.assertEquals("demo.zip", result);
+ }
+
+ @Test
+ public void testExtractDefaultFallback() {
+ String result = FilenameExtractor.extract(null,
"http://example.com/api/download";);
+ Assertions.assertTrue(result.startsWith("response-body-"));
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
index 30eca3211b..0407e9d203 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
@@ -365,6 +365,10 @@ public class HttpIT extends TestSuiteBase implements
TestResource {
// http airtable source
Container.ExecResult execResult22 =
container.executeJob("/airtable_json_to_assert.conf");
Assertions.assertEquals(0, execResult22.getExitCode());
+
+ // http binary download
+ Container.ExecResult execResult23 =
container.executeJob("/http_binary_to_assert.conf");
+ Assertions.assertEquals(0, execResult23.getExitCode());
}
@TestTemplate
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/http_binary_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/http_binary_to_assert.conf
new file mode 100644
index 0000000000..13316adeba
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/http_binary_to_assert.conf
@@ -0,0 +1,55 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Http {
+ plugin_output = "http"
+ url = "http://mockserver:1080/example/binary";
+ method = "GET"
+ format = "binary"
+ schema = {
+ fields {
+ data = bytes
+ relativePath = string
+ partIndex = long
+ }
+ }
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "http"
+ rules {
+ row_rules = [
+ {
+ rule_type = MAX_ROW
+ rule_value = 1
+ },
+ {
+ rule_type = MIN_ROW
+ rule_value = 1
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json
index f948a23d2d..6d7a40b199 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/mockserver-config.json
@@ -4943,5 +4943,21 @@
]
}
}
+ },
+ {
+ "httpRequest": {
+ "method": "GET",
+ "path": "/example/binary"
+ },
+ "httpResponse": {
+ "statusCode": 200,
+ "headers": {
+ "Content-Disposition": "attachment; filename=\"test.pdf\""
+ },
+ "body": {
+ "type": "BINARY",
+ "base64Bytes": "SGVsbG8gV29ybGQgLSBiaW5hcnkgdGVzdCBjb250ZW50"
+ }
+ }
}
]
\ No newline at end of file
