JingsongLi commented on code in PR #9271:
URL: https://github.com/apache/paimon/pull/9271#discussion_r3800554268
##########
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:
[P1] Do not retain the initial Content-Length when the replay length is
unknown.\n\nIf the initial response declares length `L` and truncates after a
prefix, while the replay is a longer chunked `200`, then `replayedLength == -1`
and this condition leaves the old `contentLength` unchanged.
`currentResponseEndExclusive` is consequently set to `L`, so `read()` returns
EOF at `L` without checking whether the replay response actually ended. The
BLOB writer can then commit truncated data instead of failing. I reproduced
this with a 100-byte initial response and a 120-byte chunked replay; the stream
returned only 100 bytes.\n\nAfter the prefix has been verified, derive the
boundary solely from the replay response, including resetting it to unknown and
reading that response to its real EOF, or fail closed.
##########
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:
[P2] Preserve compatibility with encoded-only origins.\n\n`newBodyGet`
forces `Accept-Encoding: identity`, but this status check rejects a legitimate
`406` before `openContentDecodedResponse()` can run. An origin that accepts a
normal gzip request but cannot serve identity content worked with the previous
transparent-decompression path and now always fails. If the origin ignores
`identity` and returns gzip, the current fallback also discards an already
successful response and performs a second GET, which adds request load and can
break one-shot download URLs.\n\nPlease fall back to the ordinary decoded
request when identity negotiation is rejected, or decode the first accepted
encoded response and mark that stream as non-resumable.
##########
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:
[P2] Only use Last-Modified as `If-Range` when it is demonstrably
strong.\n\n[RFC 9110 section
13.1.5](https://www.rfc-editor.org/rfc/rfc9110.html#section-13.1.5) says a
client MUST NOT generate a date-valued `If-Range` unless that date is a strong
validator under section 8.8.2.2. This code promotes every `Last-Modified` value
without checking `Date` or the one-second strength criteria. A compliant server
can therefore ignore the range and make recovery fail; a coarse timestamp
implementation can accept a same-second update and let the stream splice two
representations.\n\nUse only a strong ETag for range continuation unless the
date strength can be proven; otherwise use the full-replay and
prefix-verification path.
--
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]
