wwj6591812 commented on code in PR #9271:
URL: https://github.com/apache/paimon/pull/9271#discussion_r3803188279
##########
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) {
Review Comment:
Thanks for catching this. Fixed in `64a5ffe`: after the replayed prefix is
verified, the replay response now becomes authoritative for both
`contentLength` and `currentResponseEndExclusive`. An unknown-length chunked
replay is therefore read to its actual EOF instead of retaining the initial
response's stale length. I added
`testGetAsInputStreamReadsChunkedReplayPastInitialContentLength`, which
reproduces the 100-byte initial / 120-byte chunked replay case and asserts the
full 120 bytes.
--
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]
