wwj6591812 commented on code in PR #9271:
URL: https://github.com/apache/paimon/pull/9271#discussion_r3803190717
##########
paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java:
##########
@@ -250,4 +254,505 @@ private static <T> T newRequest(String uri,
Function<String, T> constructor) {
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 stable resource validator 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 validator;
+ 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 (validator == 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_OK) {
Review Comment:
Good catch. Fixed in `64a5ffe`: transparent decompression remains enabled on
the identity-preferred request, so a successful gzip/deflate response is
decoded and consumed from that same response and is marked non-resumable; no
second GET is issued. If identity negotiation returns HTTP 406, the rejected
response is closed and one ordinary decoded GET is attempted. I added one-shot
gzip, 406 fallback, and truncated-encoded fail-closed regression tests.
##########
paimon-api/src/main/java/org/apache/paimon/rest/HttpClientUtils.java:
##########
@@ -250,4 +254,505 @@ private static <T> T newRequest(String uri,
Function<String, T> constructor) {
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 stable resource validator 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 validator;
+ 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 (validator == 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_OK) {
+ throw httpError(newResponse.getCode());
+ }
+ if (!isIdentityEncoded(newResponse)) {
+ closeQuietly(newResponse);
+ accepted = true;
+ openContentDecodedResponse();
+ return;
+ }
+
+ HttpEntity entity = requireEntity(newResponse);
+ response = newResponse;
+ stream = entity.getContent();
+ contentLength = entity.getContentLength();
+ currentResponseEndExclusive = contentLength < 0 ?
Long.MAX_VALUE : contentLength;
+ validator = responseValidator(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 (validator == null) {
+ replayFromStart();
+ return;
+ }
+
+ HttpGet request = newBodyGet(uri);
+ request.addHeader(HttpHeaders.RANGE, "bytes=" + position + "-");
+ if (validator != null) {
+ request.addHeader(HttpHeaders.IF_RANGE, validator);
+ }
+
+ 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;
+ }
+
+ long rangeLength = range.end - range.start + 1;
+ if (!isIdentityEncoded(newResponse)) {
+ throw readFailure("range response uses a content
encoding");
+ }
+ HttpEntity entity = requireEntity(newResponse);
+ if (entity.getContentLength() >= 0 &&
entity.getContentLength() != rangeLength) {
+ throw readFailure("range response length does not match
Content-Range");
+ }
+
+ if (validator != null && hasDifferentValidator(newResponse,
validator)) {
+ throw readFailure("resource validator changed while
resuming");
+ }
+
+ response = newResponse;
+ stream = entity.getContent();
+ currentResponseEndExclusive = range.end + 1;
+ accepted = true;
+ } finally {
+ if (!accepted) {
+ closeQuietly(newResponse);
+ }
+ }
+ }
+
+ /**
+ * Replays a response without a resource validator 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");
+ }
+ if (!isIdentityEncoded(newResponse)) {
+ throw readFailure("replayed response uses a content
encoding");
+ }
+
+ HttpEntity entity = requireEntity(newResponse);
+ 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);
+ if (contentLength < 0 && replayedLength >= 0) {
+ contentLength = replayedLength;
+ }
+ response = newResponse;
+ stream = newStream;
+ currentResponseEndExclusive =
+ contentLength < 0 ? Long.MAX_VALUE : contentLength;
+ validator = responseValidator(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;
+ validator = 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");
+ request.setConfig(
+ RequestConfig.copy(DEFAULT_REQUEST_CONFIG)
+ .setContentCompressionEnabled(false)
+ .build());
+ return request;
+ }
+
+ private static String responseValidator(CloseableHttpResponse response) {
+ Header etag = response.getFirstHeader(HttpHeaders.ETAG);
+ if (etag != null) {
+ String value = etag.getValue();
+ if (value != null &&
!value.trim().toUpperCase(Locale.ROOT).startsWith("W/")) {
+ return value;
+ }
+ }
+ Header lastModified =
response.getFirstHeader(HttpHeaders.LAST_MODIFIED);
+ return lastModified == null ? null : lastModified.getValue();
Review Comment:
Agreed. Fixed in `64a5ffe`: only a syntactically strong ETag is now eligible
for `If-Range`. Last-Modified-only and weak-ETag responses use the full HTTP
200 replay plus SHA-256 prefix-verification path instead. The tests assert that
neither `Range` nor `If-Range` is sent for those two cases, while the
strong-ETag case still uses validated range continuation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
