This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4327e4e25f [api] Resume truncated HTTP GET response bodies (#9271)
4327e4e25f is described below
commit 4327e4e25f49ac78842be211da124f1fae15be27
Author: wangwj <[email protected]>
AuthorDate: Thu Aug 20 10:13:15 2026 +0800
[api] Resume truncated HTTP GET response bodies (#9271)
---
.../org/apache/paimon/rest/HttpClientUtils.java | 537 ++++++++++++++++++++-
.../apache/paimon/rest/HttpClientUtilsTest.java | 519 ++++++++++++++++++++
.../paimon/flink/HttpBlobBodyResumeITCase.java | 188 ++++++++
3 files changed, 1227 insertions(+), 17 deletions(-)
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
index e4334e60e5..1444c8118d 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java
@@ -27,6 +27,7 @@ import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.classic.methods.HttpHead;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.entity.DecompressingEntity;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
@@ -36,18 +37,36 @@ import
org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
import org.apache.hc.client5.http.ssl.HttpsSupport;
import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.ConnectionClosedException;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HttpEntity;
+import org.apache.hc.core5.http.HttpHeaders;
import org.apache.hc.core5.http.HttpStatus;
+import org.apache.hc.core5.http.TruncatedChunkException;
import org.apache.hc.core5.reactor.ssl.SSLBufferMode;
import org.apache.hc.core5.ssl.SSLContexts;
import org.apache.hc.core5.util.Timeout;
import java.io.IOException;
import java.io.InputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
/** Utils for {@link HttpClientBuilder}. */
public class HttpClientUtils {
+ private static final int MAX_BODY_RESUME_ATTEMPTS = 5;
+ private static final Pattern CONTENT_RANGE_PATTERN =
+ Pattern.compile("bytes\\s+(\\d+)-(\\d+)/(\\d+|\\*)",
Pattern.CASE_INSENSITIVE);
+ private static final RequestConfig DEFAULT_REQUEST_CONFIG =
+ RequestConfig.custom()
+ .setConnectionRequestTimeout(Timeout.ofMinutes(3))
+ .setResponseTimeout(Timeout.ofMinutes(3))
+ .build();
+
public static final CloseableHttpClient DEFAULT_HTTP_CLIENT =
createLoggingBuilder().build();
public static HttpClientBuilder createLoggingBuilder() {
@@ -60,12 +79,7 @@ public class HttpClientUtils {
public static HttpClientBuilder createBuilder() {
HttpClientBuilder clientBuilder = HttpClients.custom();
- RequestConfig requestConfig =
- RequestConfig.custom()
- .setConnectionRequestTimeout(Timeout.ofMinutes(3))
- .setResponseTimeout(Timeout.ofMinutes(3))
- .build();
- clientBuilder.setDefaultRequestConfig(requestConfig);
+ clientBuilder.setDefaultRequestConfig(DEFAULT_REQUEST_CONFIG);
clientBuilder.setConnectionManager(configureConnectionManager());
clientBuilder.setRetryStrategy(new
ExponentialHttpRequestRetryStrategy(5));
@@ -91,17 +105,7 @@ public class HttpClientUtils {
}
public static InputStream getAsInputStream(String uri) throws IOException {
- HttpGet httpGet = newHttpGet(uri);
- CloseableHttpResponse response = execute(httpGet, uri);
- int statusCode = response.getCode();
- if (statusCode != HttpStatus.SC_OK) {
- try {
- throw httpError(statusCode);
- } finally {
- response.close();
- }
- }
- return response.getEntity().getContent();
+ return new ResumableHttpInputStream(uri);
}
/**
@@ -250,4 +254,503 @@ public class HttpClientUtils {
private static RuntimeException httpError(int statusCode) {
return new RuntimeException("HTTP error code: " + statusCode);
}
+
+ /**
+ * An HTTP stream which resumes a prematurely closed response body from
the last byte already
+ * returned to the caller.
+ *
+ * <p>The request retry strategy only covers failures before response
headers are returned. A
+ * {@link ConnectionClosedException} or {@link TruncatedChunkException}
can instead be raised
+ * while the entity stream is consumed. Replaying the whole response would
duplicate bytes
+ * already written by the caller. A strong ETag allows a byte range
continuation. Without one, a
+ * complete response is replayed and its already-delivered prefix is
verified before reading
+ * continues.
+ */
+ private static class ResumableHttpInputStream extends InputStream {
+
+ private final String uri;
+ private final byte[] singleByte = new byte[1];
+
+ private CloseableHttpResponse response;
+ private InputStream stream;
+ private long position;
+ private long contentLength = -1L;
+ private long currentResponseEndExclusive = Long.MAX_VALUE;
+ private String strongEtag;
+ private boolean identityEncoded;
+ private int resumeAttempts;
+ private boolean closed;
+ private IOException terminalFailure;
+ private final MessageDigest deliveredDigest = sha256();
+
+ private ResumableHttpInputStream(String uri) throws IOException {
+ this.uri = uri;
+ openInitialResponse();
+ }
+
+ @Override
+ public int read() throws IOException {
+ int bytesRead = read(singleByte, 0, 1);
+ return bytesRead < 0 ? -1 : singleByte[0] & 0xff;
+ }
+
+ @Override
+ public int read(byte[] bytes, int offset, int length) throws
IOException {
+ if (closed) {
+ throw new IOException("HTTP response stream is closed.");
+ }
+ if (terminalFailure != null) {
+ throw new IOException(terminalFailure.getMessage());
+ }
+ if (bytes == null) {
+ throw new NullPointerException("bytes");
+ }
+ if (offset < 0 || length < 0 || length > bytes.length - offset) {
+ throw new IndexOutOfBoundsException();
+ }
+ if (length == 0) {
+ return 0;
+ }
+
+ while (true) {
+ try {
+ if (position == currentResponseEndExclusive) {
+ if (contentLength >= 0 && position < contentLength) {
+ resumeOrFail("range response ended before the
complete resource");
+ continue;
+ }
+ return -1;
+ }
+
+ int readLength =
+ (int)
+ Math.min(
+ length,
+ Math.min(
+ Integer.MAX_VALUE,
+
currentResponseEndExclusive - position));
+ int bytesRead = stream.read(bytes, offset, readLength);
+ if (bytesRead > 0) {
+ if (strongEtag == null) {
+ deliveredDigest.update(bytes, offset, bytesRead);
+ }
+ position += bytesRead;
+ if (contentLength >= 0 && position > contentLength) {
+ throw fail("response body exceeded its declared
length");
+ }
+ return bytesRead;
+ }
+ if (bytesRead < 0 && contentLength >= 0 && position <
contentLength) {
+ resumeOrFail("response body ended before its declared
length");
+ continue;
+ }
+ return bytesRead;
+ } catch (ConnectionClosedException | TruncatedChunkException
e) {
+ resumeOrFail("response body was closed before it was fully
consumed");
+ }
+ }
+ }
+
+ @Override
+ public int available() throws IOException {
+ if (closed) {
+ return 0;
+ }
+ if (terminalFailure != null) {
+ throw new IOException(terminalFailure.getMessage());
+ }
+ return stream == null ? 0 : stream.available();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ closed = true;
+ closeCurrentResponse();
+ }
+ }
+
+ private void openInitialResponse() throws IOException {
+ HttpGet request = newBodyGet(uri);
+ CloseableHttpResponse newResponse = execute(request, uri);
+ boolean accepted = false;
+ try {
+ if (newResponse.getCode() == HttpStatus.SC_NOT_ACCEPTABLE) {
+ closeQuietly(newResponse);
+ accepted = true;
+ openContentDecodedResponse();
+ return;
+ }
+ if (newResponse.getCode() != HttpStatus.SC_OK) {
+ throw httpError(newResponse.getCode());
+ }
+
+ HttpEntity entity = requireEntity(newResponse);
+ if (entity instanceof DecompressingEntity ||
!isIdentityEncoded(newResponse)) {
+ response = newResponse;
+ stream = entity.getContent();
+ contentLength =
+ entity instanceof DecompressingEntity ? -1L :
entity.getContentLength();
+ currentResponseEndExclusive =
+ contentLength < 0 ? Long.MAX_VALUE : contentLength;
+ strongEtag = null;
+ identityEncoded = false;
+ accepted = true;
+ return;
+ }
+
+ response = newResponse;
+ stream = entity.getContent();
+ contentLength = entity.getContentLength();
+ currentResponseEndExclusive = contentLength < 0 ?
Long.MAX_VALUE : contentLength;
+ strongEtag = responseStrongEtag(newResponse);
+ identityEncoded = true;
+ accepted = true;
+ } finally {
+ if (!accepted) {
+ closeQuietly(newResponse);
+ }
+ }
+ }
+
+ private void resumeOrFail(String reason) throws IOException {
+ try {
+ resume(reason);
+ } catch (IOException e) {
+ terminalFailure = e;
+ discardCurrentResponse();
+ throw e;
+ } catch (RuntimeException e) {
+ Integer statusCode = getHttpStatusCode(e);
+ terminalFailure =
+ readFailure(
+ statusCode == null
+ ? "response restart failed"
+ : "server returned HTTP "
+ + statusCode
+ + " while restarting the
response");
+ discardCurrentResponse();
+ throw terminalFailure;
+ }
+ }
+
+ private void resume(String reason) throws IOException {
+ if (resumeAttempts >= MAX_BODY_RESUME_ATTEMPTS) {
+ throw readFailure(
+ reason + " after " + MAX_BODY_RESUME_ATTEMPTS + "
resume attempts");
+ }
+ resumeAttempts++;
+ discardCurrentResponse();
+
+ if (position == 0) {
+ openInitialResponse();
+ return;
+ }
+ if (!identityEncoded) {
+ throw readFailure("encoded response bodies cannot be resumed
safely");
+ }
+ if (strongEtag == null) {
+ replayFromStart();
+ return;
+ }
+
+ HttpGet request = newBodyGet(uri);
+ request.addHeader(HttpHeaders.RANGE, "bytes=" + position + "-");
+ request.addHeader(HttpHeaders.IF_RANGE, strongEtag);
+
+ CloseableHttpResponse newResponse = execute(request, uri);
+ boolean accepted = false;
+ try {
+ if (newResponse.getCode() != HttpStatus.SC_PARTIAL_CONTENT) {
+ throw readFailure(
+ "server did not honor the range request (HTTP "
+ + newResponse.getCode()
+ + ")");
+ }
+
+ Range range = parseContentRange(newResponse);
+ if (range.start != position) {
+ throw readFailure(
+ "server resumed at byte " + range.start + "
instead of " + position);
+ }
+ if (contentLength >= 0 && range.total != contentLength) {
+ throw readFailure(
+ "resource length changed from " + contentLength +
" to " + range.total);
+ }
+ if (contentLength < 0) {
+ contentLength = range.total;
+ }
+
+ HttpEntity entity = requireEntity(newResponse);
+ long rangeLength = range.end - range.start + 1;
+ if (entity instanceof DecompressingEntity ||
!isIdentityEncoded(newResponse)) {
+ throw readFailure("range response uses a content
encoding");
+ }
+ if (entity.getContentLength() >= 0 &&
entity.getContentLength() != rangeLength) {
+ throw readFailure("range response length does not match
Content-Range");
+ }
+
+ if (hasDifferentStrongEtag(newResponse, strongEtag)) {
+ throw readFailure("strong ETag changed while resuming");
+ }
+
+ response = newResponse;
+ stream = entity.getContent();
+ currentResponseEndExclusive = range.end + 1;
+ accepted = true;
+ } finally {
+ if (!accepted) {
+ closeQuietly(newResponse);
+ }
+ }
+ }
+
+ /**
+ * Replays a response without a strong ETag from byte zero and
verifies that its prefix is
+ * identical to the bytes already returned. Once the prefix matches,
continuing with the
+ * same response cannot combine bytes from two different
representations.
+ */
+ private void replayFromStart() throws IOException {
+ byte[] expectedPrefixDigest = digestSnapshot(deliveredDigest);
+ while (true) {
+ HttpGet request = newBodyGet(uri);
+ CloseableHttpResponse newResponse = execute(request, uri);
+ boolean accepted = false;
+ try {
+ if (newResponse.getCode() != HttpStatus.SC_OK) {
+ throw readFailure(
+ "server returned HTTP "
+ + newResponse.getCode()
+ + " while replaying the response");
+ }
+ HttpEntity entity = requireEntity(newResponse);
+ if (entity instanceof DecompressingEntity ||
!isIdentityEncoded(newResponse)) {
+ throw readFailure("replayed response uses a content
encoding");
+ }
+ long replayedLength = entity.getContentLength();
+ if (contentLength >= 0
+ && replayedLength >= 0
+ && contentLength != replayedLength) {
+ throw readFailure(
+ "resource length changed from "
+ + contentLength
+ + " to "
+ + replayedLength);
+ }
+
+ InputStream newStream = entity.getContent();
+ verifyReplayedPrefix(newStream, expectedPrefixDigest);
+ contentLength = replayedLength;
+ response = newResponse;
+ stream = newStream;
+ currentResponseEndExclusive =
+ replayedLength < 0 ? Long.MAX_VALUE :
replayedLength;
+ strongEtag = responseStrongEtag(newResponse);
+ identityEncoded = true;
+ accepted = true;
+ return;
+ } catch (ConnectionClosedException | TruncatedChunkException
e) {
+ if (resumeAttempts >= MAX_BODY_RESUME_ATTEMPTS) {
+ throw readFailure(
+ "response replay failed after "
+ + MAX_BODY_RESUME_ATTEMPTS
+ + " resume attempts");
+ }
+ resumeAttempts++;
+ } finally {
+ if (!accepted) {
+ closeQuietly(newResponse);
+ }
+ }
+ }
+ }
+
+ private void verifyReplayedPrefix(InputStream newStream, byte[]
expectedPrefixDigest)
+ throws IOException {
+ MessageDigest replayedDigest = sha256();
+ byte[] buffer = new byte[8192];
+ long remaining = position;
+ while (remaining > 0) {
+ int bytesRead = newStream.read(buffer, 0, (int)
Math.min(buffer.length, remaining));
+ if (bytesRead < 0) {
+ throw new ConnectionClosedException(
+ "Response ended before the previously delivered
prefix.");
+ }
+ if (bytesRead == 0) {
+ throw new IOException("HTTP response returned zero bytes
while replaying.");
+ }
+ replayedDigest.update(buffer, 0, bytesRead);
+ remaining -= bytesRead;
+ }
+ if (!MessageDigest.isEqual(expectedPrefixDigest,
replayedDigest.digest())) {
+ throw readFailure("resource content changed while replaying
the response");
+ }
+ }
+
+ /**
+ * Preserves the old transparent content-decoding behavior for a
server which ignores the
+ * identity request. Encoded response bodies cannot use byte-offset
recovery because their
+ * decoded positions do not match wire byte ranges.
+ */
+ private void openContentDecodedResponse() throws IOException {
+ HttpGet request = newHttpGet(uri);
+ CloseableHttpResponse newResponse = execute(request, uri);
+ boolean accepted = false;
+ try {
+ if (newResponse.getCode() != HttpStatus.SC_OK) {
+ throw httpError(newResponse.getCode());
+ }
+
+ HttpEntity entity = requireEntity(newResponse);
+ response = newResponse;
+ stream = entity.getContent();
+ contentLength = -1L;
+ currentResponseEndExclusive = Long.MAX_VALUE;
+ strongEtag = null;
+ identityEncoded = false;
+ accepted = true;
+ } finally {
+ if (!accepted) {
+ closeQuietly(newResponse);
+ }
+ }
+ }
+
+ private void closeCurrentResponse() throws IOException {
+ stream = null;
+ if (response != null) {
+ try {
+ response.close();
+ } finally {
+ response = null;
+ }
+ }
+ }
+
+ private void discardCurrentResponse() {
+ try {
+ closeCurrentResponse();
+ } catch (IOException ignored) {
+ // The response is being discarded precisely because its body
is incomplete.
+ }
+ }
+
+ private IOException readFailure(String reason) {
+ return new IOException(
+ "Failed to resume HTTP response for uri: "
+ + SensitiveConfigUtils.sanitizeUri(uri)
+ + "; position="
+ + position
+ + ", contentLength="
+ + contentLength
+ + ", recoveryAttempts="
+ + resumeAttempts
+ + "; "
+ + reason);
+ }
+
+ private IOException fail(String reason) {
+ terminalFailure = readFailure(reason);
+ discardCurrentResponse();
+ return terminalFailure;
+ }
+ }
+
+ private static HttpEntity requireEntity(CloseableHttpResponse response)
throws IOException {
+ HttpEntity entity = response.getEntity();
+ if (entity == null) {
+ throw new IOException("HTTP response has no entity.");
+ }
+ return entity;
+ }
+
+ private static HttpGet newBodyGet(String uri) {
+ HttpGet request = newHttpGet(uri);
+ request.addHeader(HttpHeaders.ACCEPT_ENCODING, "identity");
+ return request;
+ }
+
+ private static String responseStrongEtag(CloseableHttpResponse response) {
+ Header etag = response.getFirstHeader(HttpHeaders.ETAG);
+ if (etag == null || etag.getValue() == null) {
+ return null;
+ }
+
+ String value = etag.getValue().trim();
+ return value.length() >= 2
+ && value.charAt(0) == '"'
+ && value.charAt(value.length() - 1) == '"'
+ && value.indexOf('"', 1) == value.length() - 1
+ ? value
+ : null;
+ }
+
+ private static boolean hasDifferentStrongEtag(
+ CloseableHttpResponse response, String expectedStrongEtag) {
+ Header etag = response.getFirstHeader(HttpHeaders.ETAG);
+ return etag != null &&
!expectedStrongEtag.equals(responseStrongEtag(response));
+ }
+
+ private static boolean isIdentityEncoded(CloseableHttpResponse response) {
+ Header contentEncoding =
response.getFirstHeader(HttpHeaders.CONTENT_ENCODING);
+ return contentEncoding == null
+ ||
"identity".equalsIgnoreCase(contentEncoding.getValue().trim());
+ }
+
+ private static void closeQuietly(CloseableHttpResponse response) {
+ try {
+ response.close();
+ } catch (IOException ignored) {
+ // Best effort cleanup while preserving the original failure.
+ }
+ }
+
+ private static MessageDigest sha256() {
+ try {
+ return MessageDigest.getInstance("SHA-256");
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is not available.", e);
+ }
+ }
+
+ private static byte[] digestSnapshot(MessageDigest digest) throws
IOException {
+ try {
+ return ((MessageDigest) digest.clone()).digest();
+ } catch (CloneNotSupportedException e) {
+ throw new IOException("SHA-256 digest cannot be cloned.");
+ }
+ }
+
+ private static Range parseContentRange(CloseableHttpResponse response)
throws IOException {
+ Header header = response.getFirstHeader(HttpHeaders.CONTENT_RANGE);
+ Matcher matcher =
+ header == null ? null :
CONTENT_RANGE_PATTERN.matcher(header.getValue().trim());
+ if (matcher == null || !matcher.matches() ||
"*".equals(matcher.group(3))) {
+ throw new IOException("Invalid Content-Range in HTTP resume
response.");
+ }
+ try {
+ long start = Long.parseLong(matcher.group(1));
+ long end = Long.parseLong(matcher.group(2));
+ long total = Long.parseLong(matcher.group(3));
+ if (start < 0 || end < start || total <= end) {
+ throw new IOException("Invalid Content-Range in HTTP resume
response.");
+ }
+ return new Range(start, end, total);
+ } catch (NumberFormatException e) {
+ throw new IOException("Invalid Content-Range in HTTP resume
response.");
+ }
+ }
+
+ private static class Range {
+
+ private final long start;
+ private final long end;
+ private final long total;
+
+ private Range(long start, long end, long total) {
+ this.start = start;
+ this.end = end;
+ this.total = total;
+ }
+ }
}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
index cb56cc1fae..5b1c3c2700 100644
--- a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
+++ b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientUtilsTest.java
@@ -28,10 +28,16 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPOutputStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -250,6 +256,468 @@ public class HttpClientUtilsTest {
}
}
+ @Test
+ public void testGetAsInputStreamPreservesContentDecoding() throws
Exception {
+ byte[] payload = payload(4096);
+ byte[] compressed = gzip(payload);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/encoded",
+ exchange -> {
+ if (requestCount.incrementAndGet() > 1) {
+ respond(exchange, 410, new byte[0]);
+ return;
+ }
+ exchange.getResponseHeaders().add("Content-Encoding",
"gzip");
+ respond(exchange, 200, compressed);
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/encoded"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(1);
+ }
+
+ @Test
+ public void testGetAsInputStreamFallsBackWhenIdentityEncodingIsRejected()
throws Exception {
+ byte[] payload = payload(4096);
+ byte[] compressed = gzip(payload);
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> firstAcceptEncoding = new AtomicReference<>();
+ AtomicReference<String> secondAcceptEncoding = new AtomicReference<>();
+ registerHandler(
+ "/encoded-only",
+ exchange -> {
+ int currentRequest = requestCount.incrementAndGet();
+ String acceptEncoding =
+
exchange.getRequestHeaders().getFirst("Accept-Encoding");
+ if (currentRequest == 1) {
+ firstAcceptEncoding.set(acceptEncoding);
+ respond(exchange, 406, new byte[0]);
+ return;
+ }
+
+ secondAcceptEncoding.set(acceptEncoding);
+ exchange.getResponseHeaders().add("Content-Encoding",
"gzip");
+ respond(exchange, 200, compressed);
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/encoded-only"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(2);
+ assertThat(firstAcceptEncoding).hasValue("identity");
+ assertThat(secondAcceptEncoding.get()).contains("gzip");
+ }
+
+ @Test
+ public void testGetAsInputStreamDoesNotResumeTruncatedEncodedResponse()
throws Exception {
+ byte[] compressed = gzip(payload(4096));
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/truncated-encoded",
+ exchange -> {
+ requestCount.incrementAndGet();
+ exchange.getResponseHeaders().add("Content-Encoding",
"gzip");
+ respondTruncated(
+ exchange, 200, compressed.length, compressed,
compressed.length / 2);
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/truncated-encoded"))) {
+ assertThatThrownBy(() -> readAll(in))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("encoded response bodies cannot be
resumed safely");
+ }
+ assertThat(requestCount).hasValue(1);
+ }
+
+ @Test
+ public void testGetAsInputStreamResumesTruncatedResponseBody() throws
Exception {
+ byte[] payload = payload(128 * 1024);
+ int truncatedLength = 89_075;
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> ifRange = new AtomicReference<>();
+ AtomicReference<String> acceptEncoding = new AtomicReference<>();
+ registerHandler(
+ "/truncated",
+ exchange -> {
+ int currentRequest = requestCount.incrementAndGet();
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (currentRequest == 1) {
+ respondTruncated(exchange, payload, truncatedLength);
+ return;
+ }
+
+ range.set(exchange.getRequestHeaders().getFirst("Range"));
+
ifRange.set(exchange.getRequestHeaders().getFirst("If-Range"));
+
acceptEncoding.set(exchange.getRequestHeaders().getFirst("Accept-Encoding"));
+ byte[] remaining = Arrays.copyOfRange(payload,
truncatedLength, payload.length);
+ exchange.getResponseHeaders()
+ .add(
+ "Content-Range",
+ String.format(
+ "bytes %d-%d/%d",
+ truncatedLength, payload.length -
1, payload.length));
+ respond(exchange, 206, remaining);
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/truncated"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(2);
+ assertThat(range).hasValue("bytes=" + truncatedLength + "-");
+ assertThat(ifRange).hasValue("\"image-v1\"");
+ assertThat(acceptEncoding).hasValue("identity");
+ }
+
+ @Test
+ public void
testGetAsInputStreamReplaysAndVerifiesWithoutResourceValidator() throws
Exception {
+ byte[] payload = payload(4096);
+ int truncatedLength = 1024;
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> acceptEncoding = new AtomicReference<>();
+ registerHandler(
+ "/truncated-no-validator",
+ exchange -> {
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, truncatedLength);
+ return;
+ }
+
+ range.set(exchange.getRequestHeaders().getFirst("Range"));
+
acceptEncoding.set(exchange.getRequestHeaders().getFirst("Accept-Encoding"));
+ respond(exchange, 200, payload);
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/truncated-no-validator"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(2);
+ assertThat(range.get()).isNull();
+ assertThat(acceptEncoding).hasValue("identity");
+ }
+
+ @Test
+ public void
testGetAsInputStreamReadsChunkedReplayPastInitialContentLength() throws
Exception {
+ byte[] replayedPayload = payload(120);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/longer-chunked-replay",
+ exchange -> {
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, 200, 100, replayedPayload,
40);
+ } else {
+ respondChunked(exchange, 200, replayedPayload);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/longer-chunked-replay"))) {
+ assertThat(readAll(in)).isEqualTo(replayedPayload);
+ }
+ assertThat(requestCount).hasValue(2);
+ }
+
+ @Test
+ public void testGetAsInputStreamReplaysWhenOnlyLastModifiedIsAvailable()
throws Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> ifRange = new AtomicReference<>();
+ registerHandler(
+ "/last-modified-only",
+ exchange -> {
+ exchange.getResponseHeaders()
+ .add("Last-Modified", "Mon, 17 Aug 2026 00:00:00
GMT");
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 1024);
+ } else {
+
range.set(exchange.getRequestHeaders().getFirst("Range"));
+
ifRange.set(exchange.getRequestHeaders().getFirst("If-Range"));
+ respond(exchange, 200, payload);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/last-modified-only"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(2);
+ assertThat(range.get()).isNull();
+ assertThat(ifRange.get()).isNull();
+ }
+
+ @Test
+ public void testGetAsInputStreamReplaysWhenOnlyWeakEtagIsAvailable()
throws Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> ifRange = new AtomicReference<>();
+ registerHandler(
+ "/weak-etag",
+ exchange -> {
+ exchange.getResponseHeaders().add("ETag",
"W/\"image-v1\"");
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 1024);
+ } else {
+
range.set(exchange.getRequestHeaders().getFirst("Range"));
+
ifRange.set(exchange.getRequestHeaders().getFirst("If-Range"));
+ respond(exchange, 200, payload);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/weak-etag"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(2);
+ assertThat(range.get()).isNull();
+ assertThat(ifRange.get()).isNull();
+ }
+
+ @Test
+ public void
testGetAsInputStreamRejectsChangedReplayWithoutResourceValidator()
+ throws Exception {
+ byte[] payload = payload(4096);
+ byte[] changedPayload = payload.clone();
+ changedPayload[17] ^= 1;
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/changed-replay",
+ exchange -> {
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 1024);
+ } else {
+ respond(exchange, 200, changedPayload);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/changed-replay"))) {
+ assertThatThrownBy(() -> readAll(in))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("resource content changed");
+ }
+ assertThat(requestCount).hasValue(2);
+ }
+
+ @Test
+ public void testGetAsInputStreamResumesMultipleTruncatedBodies() throws
Exception {
+ byte[] payload = payload(16 * 1024);
+ int firstEnd = 2048;
+ int secondEnd = 3072;
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> firstRange = new AtomicReference<>();
+ AtomicReference<String> secondRange = new AtomicReference<>();
+ registerHandler(
+ "/multiple-truncations",
+ exchange -> {
+ int currentRequest = requestCount.incrementAndGet();
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (currentRequest == 1) {
+ respondTruncated(exchange, payload, firstEnd);
+ return;
+ }
+
+ int start = currentRequest == 2 ? firstEnd : secondEnd;
+ if (currentRequest == 2) {
+
firstRange.set(exchange.getRequestHeaders().getFirst("Range"));
+ } else {
+
secondRange.set(exchange.getRequestHeaders().getFirst("Range"));
+ }
+ exchange.getResponseHeaders()
+ .add(
+ "Content-Range",
+ String.format(
+ "bytes %d-%d/%d",
+ start, payload.length - 1,
payload.length));
+ byte[] remaining = Arrays.copyOfRange(payload, start,
payload.length);
+ if (currentRequest == 2) {
+ respondTruncated(exchange, 206, remaining.length,
remaining, 1024);
+ } else {
+ respond(exchange, 206, remaining);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/multiple-truncations"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(3);
+ assertThat(firstRange).hasValue("bytes=" + firstEnd + "-");
+ assertThat(secondRange).hasValue("bytes=" + secondEnd + "-");
+ }
+
+ @Test
+ public void testGetAsInputStreamContinuesAfterBoundedRangeResponse()
throws Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> lastRange = new AtomicReference<>();
+ registerHandler(
+ "/bounded-range",
+ exchange -> {
+ int currentRequest = requestCount.incrementAndGet();
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (currentRequest == 1) {
+ respondTruncated(exchange, payload, 1024);
+ } else if (currentRequest == 2) {
+ exchange.getResponseHeaders().add("Content-Range",
"bytes 1024-2047/4096");
+ respond(exchange, 206, Arrays.copyOfRange(payload,
1024, 2048));
+ } else {
+
lastRange.set(exchange.getRequestHeaders().getFirst("Range"));
+ exchange.getResponseHeaders().add("Content-Range",
"bytes 2048-4095/4096");
+ respond(exchange, 206, Arrays.copyOfRange(payload,
2048, 4096));
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/bounded-range"))) {
+ assertThat(readAll(in)).isEqualTo(payload);
+ }
+ assertThat(requestCount).hasValue(3);
+ assertThat(lastRange).hasValue("bytes=2048-");
+ }
+
+ @Test
+ public void testGetAsInputStreamFailsWhenServerIgnoresRange() throws
Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/ignore-range",
+ exchange -> {
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 1024);
+ } else {
+ respond(exchange, 200, payload);
+ }
+ });
+
+ String uri = url("/ignore-range") + "?sig=BODY_RESUME_SECRET";
+ try (InputStream in = HttpClientUtils.getAsInputStream(uri)) {
+ assertThatThrownBy(() -> readAll(in))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("did not honor the range request")
+ .satisfies(
+ e -> {
+
assertThat(String.valueOf(e)).doesNotContain("BODY_RESUME_SECRET");
+
assertThat(e.getMessage()).doesNotContain("sig=");
+ });
+ }
+ assertThat(requestCount).hasValue(2);
+ }
+
+ @Test
+ public void testGetAsInputStreamFailsForMismatchedContentRange() throws
Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/wrong-content-range",
+ exchange -> {
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 1024);
+ return;
+ }
+
+ exchange.getResponseHeaders().add("Content-Range", "bytes
1023-4095/4096");
+ respond(exchange, 206, Arrays.copyOfRange(payload, 1023,
payload.length));
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/wrong-content-range"))) {
+ assertThatThrownBy(() -> readAll(in))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("resumed at byte 1023 instead of
1024");
+ }
+ assertThat(requestCount).hasValue(2);
+ }
+
+ @Test
+ public void testGetAsInputStreamFailsAfterBoundedResumeAttempts() throws
Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/resume-exhausted",
+ exchange -> {
+ int currentRequest = requestCount.incrementAndGet();
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (currentRequest == 1) {
+ respondTruncated(exchange, payload, 1024);
+ return;
+ }
+
+ int start = 1024 + currentRequest - 2;
+ exchange.getResponseHeaders()
+ .add(
+ "Content-Range",
+ String.format(
+ "bytes %d-%d/%d",
+ start, payload.length - 1,
payload.length));
+ byte[] remaining = Arrays.copyOfRange(payload, start,
payload.length);
+ respondTruncated(exchange, 206, remaining.length,
remaining, 1);
+ });
+
+ String uri = url("/resume-exhausted") + "?sig=BODY_RESUME_SECRET";
+ try (InputStream in = HttpClientUtils.getAsInputStream(uri)) {
+ assertThatThrownBy(() -> readAll(in))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("after 5 resume attempts")
+ .hasMessageContaining(url("/resume-exhausted"))
+ .hasMessageContaining("position=1029")
+ .hasMessageContaining("contentLength=4096")
+ .hasMessageContaining("recoveryAttempts=5")
+ .satisfies(
+ e -> {
+
assertThat(String.valueOf(e)).doesNotContain("BODY_RESUME_SECRET");
+
assertThat(e.getMessage()).doesNotContain("sig=");
+ });
+ assertThatThrownBy(in::read)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("after 5 resume attempts");
+ }
+ assertThat(requestCount).hasValue(6);
+ }
+
+ @Test
+ public void testGetAsInputStreamDoesNotResumeAfterClose() throws Exception
{
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/closed-stream",
+ exchange -> {
+ requestCount.incrementAndGet();
+ respond(exchange, 200, payload(4096));
+ });
+
+ InputStream in =
HttpClientUtils.getAsInputStream(url("/closed-stream"));
+ assertThat(in.read()).isNotNegative();
+ in.close();
+ assertThatThrownBy(in::read)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("stream is closed");
+ assertThat(requestCount).hasValue(1);
+ }
+
+ @Test
+ public void testGetAsInputStreamKeepsRestartStatusFailureTerminal() throws
Exception {
+ byte[] payload = payload(4096);
+ AtomicInteger requestCount = new AtomicInteger();
+ registerHandler(
+ "/restart-status",
+ exchange -> {
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, 0);
+ } else {
+ respond(exchange, 404, new byte[0]);
+ }
+ });
+
+ try (InputStream in =
HttpClientUtils.getAsInputStream(url("/restart-status"))) {
+ assertThatThrownBy(in::read)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("server returned HTTP 404");
+ assertThatThrownBy(in::read)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("server returned HTTP 404");
+ }
+ assertThat(requestCount).hasValue(2);
+ }
+
@Test
public void testIsNotFoundError() {
RuntimeException exception =
@@ -330,4 +798,55 @@ public class HttpClientUtilsTest {
exchange.close();
}
}
+
+ private static void respondTruncated(HttpExchange exchange, byte[] body,
int truncatedLength)
+ throws IOException {
+ respondTruncated(exchange, 200, body.length, body, truncatedLength);
+ }
+
+ private static void respondTruncated(
+ HttpExchange exchange,
+ int statusCode,
+ long declaredLength,
+ byte[] body,
+ int truncatedLength)
+ throws IOException {
+ exchange.sendResponseHeaders(statusCode, declaredLength);
+ OutputStream outputStream = exchange.getResponseBody();
+ outputStream.write(body, 0, truncatedLength);
+ outputStream.flush();
+ exchange.close();
+ }
+
+ private static void respondChunked(HttpExchange exchange, int statusCode,
byte[] body)
+ throws IOException {
+ exchange.sendResponseHeaders(statusCode, 0);
+ try (OutputStream outputStream = exchange.getResponseBody()) {
+ outputStream.write(body);
+ }
+ }
+
+ private static byte[] payload(int length) {
+ byte[] payload = new byte[length];
+ new Random(20260817L).nextBytes(payload);
+ return payload;
+ }
+
+ private static byte[] gzip(byte[] payload) throws IOException {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try (GZIPOutputStream gzipOutputStream = new
GZIPOutputStream(outputStream)) {
+ gzipOutputStream.write(payload);
+ }
+ return outputStream.toByteArray();
+ }
+
+ private static byte[] readAll(InputStream inputStream) throws IOException {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ byte[] buffer = new byte[8192];
+ int bytesRead;
+ while ((bytesRead = inputStream.read(buffer)) >= 0) {
+ outputStream.write(buffer, 0, bytesRead);
+ }
+ return outputStream.toByteArray();
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/HttpBlobBodyResumeITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/HttpBlobBodyResumeITCase.java
new file mode 100644
index 0000000000..2a5c5350d2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/HttpBlobBodyResumeITCase.java
@@ -0,0 +1,188 @@
+/*
+ * 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.paimon.flink;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for resuming truncated HTTP response bodies while writing
BLOBs. */
+public class HttpBlobBodyResumeITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testHttpBlobWriteResumesWithStrongEtag() throws Exception {
+ byte[] payload = payload(128 * 1024);
+ int truncatedLength = 89_075;
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> ifRange = new AtomicReference<>();
+
+ HttpServer server = newServer();
+ server.createContext(
+ "/strong-etag",
+ exchange -> {
+ exchange.getResponseHeaders().add("ETag", "\"image-v1\"");
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, truncatedLength);
+ return;
+ }
+
+ range.set(exchange.getRequestHeaders().getFirst("Range"));
+
ifRange.set(exchange.getRequestHeaders().getFirst("If-Range"));
+ byte[] remaining = Arrays.copyOfRange(payload,
truncatedLength, payload.length);
+ exchange.getResponseHeaders()
+ .add(
+ "Content-Range",
+ String.format(
+ "bytes %d-%d/%d",
+ truncatedLength, payload.length -
1, payload.length));
+ respond(exchange, 206, remaining);
+ });
+ server.start();
+
+ try {
+ String url = url(server, "/strong-etag");
+ // A successful recovery must preserve the BLOB even when terminal
fetch failures are
+ // configured to fall back to NULL.
+ createBlobTable("strong_etag_blob_table", true);
+ batchSql(
+ "INSERT INTO strong_etag_blob_table VALUES"
+ + " (1, sys.path_to_descriptor('"
+ + url
+ + "'))");
+
+ assertBlobEquals("strong_etag_blob_table", payload);
+ assertThat(requestCount).hasValue(2);
+ assertThat(range).hasValue("bytes=" + truncatedLength + "-");
+ assertThat(ifRange).hasValue("\"image-v1\"");
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testHttpBlobWriteReplaysWithoutStrongEtag() throws Exception {
+ byte[] payload = payload(32 * 1024);
+ int truncatedLength = 7_321;
+ AtomicInteger requestCount = new AtomicInteger();
+ AtomicReference<String> range = new AtomicReference<>();
+ AtomicReference<String> ifRange = new AtomicReference<>();
+
+ HttpServer server = newServer();
+ server.createContext(
+ "/no-strong-etag",
+ exchange -> {
+ if (requestCount.incrementAndGet() == 1) {
+ respondTruncated(exchange, payload, truncatedLength);
+ return;
+ }
+
+ range.set(exchange.getRequestHeaders().getFirst("Range"));
+
ifRange.set(exchange.getRequestHeaders().getFirst("If-Range"));
+ respond(exchange, 200, payload);
+ });
+ server.start();
+
+ try {
+ String url = url(server, "/no-strong-etag");
+ createBlobTable("replayed_blob_table", false);
+ batchSql(
+ "INSERT INTO replayed_blob_table VALUES"
+ + " (1, sys.path_to_descriptor('"
+ + url
+ + "'))");
+
+ assertBlobEquals("replayed_blob_table", payload);
+ assertThat(requestCount).hasValue(2);
+ assertThat(range.get()).isNull();
+ assertThat(ifRange.get()).isNull();
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private void createBlobTable(String tableName, boolean
writeNullOnFetchFailure) {
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE %s (id INT, picture BYTES) WITH ("
+ + "'row-tracking.enabled'='true',"
+ + "'data-evolution.enabled'='true',"
+ + "'blob-field'='picture',"
+ + "'blob-as-descriptor'='true'%s)",
+ tableName,
+ writeNullOnFetchFailure
+ ? ",'blob-write-null-on-fetch-failure'='true'"
+ : ""));
+ }
+
+ private void assertBlobEquals(String tableName, byte[] expected) {
+ batchSql("ALTER TABLE %s SET ('blob-as-descriptor'='false')",
tableName);
+ List<Row> rows = batchSql("SELECT id, picture FROM %s", tableName);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getField(0)).isEqualTo(1);
+ assertThat((byte[]) rows.get(0).getField(1)).isEqualTo(expected);
+ assertThat(findLatestSnapshot(tableName)).isNotNull();
+ }
+
+ private static HttpServer newServer() throws IOException {
+ return HttpServer.create(new InetSocketAddress(0), 0);
+ }
+
+ private static String url(HttpServer server, String path) {
+ return String.format("http://localhost:%d%s";,
server.getAddress().getPort(), path);
+ }
+
+ private static void respondTruncated(HttpExchange exchange, byte[]
payload, int truncatedLength)
+ throws IOException {
+ exchange.sendResponseHeaders(200, payload.length);
+ OutputStream out = exchange.getResponseBody();
+ out.write(payload, 0, truncatedLength);
+ out.flush();
+ exchange.close();
+ }
+
+ private static void respond(HttpExchange exchange, int statusCode, byte[]
payload)
+ throws IOException {
+ exchange.sendResponseHeaders(statusCode, payload.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(payload);
+ }
+ }
+
+ private static byte[] payload(int length) {
+ byte[] payload = new byte[length];
+ byte[] seed =
"paimon-http-blob-resume".getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i < payload.length; i++) {
+ payload[i] = (byte) (seed[i % seed.length] ^ (i * 31) ^ (i >>> 8));
+ }
+ return payload;
+ }
+}
