From 9acd393741389403c0883cb127a4d5653e1f1419 Mon Sep 17 00:00:00 2001 From: Animesh Sahu Date: Tue, 9 Jun 2026 11:19:03 +0530 Subject: [PATCH 1/3] Cleanly close the reach channel in the GoogleHadoopFSInputStream so that read to EOF does not happen on calling close() unnecessarily --- .../gcsio/GoogleCloudStorageReadChannel.java | 59 +++++++-- .../GoogleCloudStorageReadChannelTest.java | 114 ++++++++++++++++++ 2 files changed, 163 insertions(+), 10 deletions(-) diff --git a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java index eae3e1544..9b9c2aac9 100644 --- a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java +++ b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java @@ -147,6 +147,16 @@ public class GoogleCloudStorageReadChannel implements SeekableByteChannel { // 3. Test that footer prefetch always disabled for gzipped files. private byte[] footerContent; + /** + * The {@link HttpResponse} for the media stream currently returned from {@link #openStream}, if + * any. Held so that {@link #closeContentChannel} can call {@link HttpResponse#disconnect()} + * before closing the {@link #contentChannel}. Otherwise Apache HTTP's {@code + * ContentLengthInputStream} drains the entire remaining entity on {@code InputStream#close()}, + * which can take minutes on a large object when the caller closes after only reading a small + * prefix (e.g. Hadoop distcp aborting a copy). + */ + @Nullable private HttpResponse openMediaResponse; + @VisibleForTesting protected boolean metadataInitialized = false; /** @@ -525,6 +535,7 @@ public boolean isOpen() { * already responsible for performing local cleanup at the time the exception was raised. */ protected void closeContentChannel() { + disconnectOpenMediaResponse(); if (contentChannel != null) { logger.atFiner().log("Closing internal contentChannel for '%s'", resourceId); try { @@ -541,6 +552,33 @@ protected void closeContentChannel() { } } + /** + * Aborts the underlying HTTP request for the current media body, if any. Must run before closing + * the {@link #contentChannel} so that closing the Apache {@code ContentLengthInputStream} does + * not synchronously read the rest of the response body. + */ + void disconnectHttpResponse(@Nullable HttpResponse response) { + if (response == null) { + return; + } + try { + response.disconnect(); + } catch (Exception e) { + GoogleCloudStorageEventBus.postOnException(); + logger.atFine().withCause(e).log( + "Got an exception on HttpResponse.disconnect() for '%s'; ignoring it.", resourceId); + } + } + + private void disconnectOpenMediaResponse() { + if (openMediaResponse == null) { + return; + } + HttpResponse response = openMediaResponse; + openMediaResponse = null; + disconnectHttpResponse(response); + } + private void resetContentChannel() { checkState(contentChannel == null, "contentChannel should be null for '%s'", resourceId); contentChannelPosition = -1; @@ -1020,6 +1058,7 @@ protected InputStream openStream(long bytesToRead) throws IOException { metadataInitialized, "metadata should be initialized already for '%s'", resourceId); if (size == 0) { resetContentChannel(); + disconnectHttpResponse(response); return new ByteArrayInputStream(new byte[0]); } if (gzipEncoded) { @@ -1030,6 +1069,7 @@ protected InputStream openStream(long bytesToRead) throws IOException { contentChannelEnd = size; } else { resetContentChannel(); + disconnectHttpResponse(response); return openStream(bytesToRead); } } @@ -1118,18 +1158,17 @@ protected InputStream openStream(long bytesToRead) throws IOException { currentPosition, resourceId); - return new GcsReadDurationTrackerStream( - contentStream, - UriPaths.fromResourceId(resourceId, /* allowEmptyObjectName= */ false), - response.getHeaders(), - readOptions.getLatencyLoggingThreshold()); + GcsReadDurationTrackerStream tracked = + new GcsReadDurationTrackerStream( + contentStream, + UriPaths.fromResourceId(resourceId, /* allowEmptyObjectName= */ false), + response.getHeaders(), + readOptions.getLatencyLoggingThreshold()); + openMediaResponse = response; + return tracked; } catch (IOException e) { GoogleCloudStorageEventBus.postOnException(); - try { - response.disconnect(); - } catch (IOException closeException) { - e.addSuppressed(closeException); - } + disconnectHttpResponse(response); throw e; } } diff --git a/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java b/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java index a37de7ad4..07c84cfa8 100644 --- a/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java +++ b/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java @@ -37,6 +37,7 @@ import static org.junit.Assert.assertThrows; import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpResponse; import com.google.api.client.json.gson.GsonFactory; import com.google.api.client.testing.http.MockHttpTransport; import com.google.api.client.testing.http.MockLowLevelHttpResponse; @@ -63,6 +64,7 @@ import java.util.List; import java.util.Random; import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -1060,6 +1062,118 @@ public void readBeyondChannelLength() throws Exception { assertThat(readChannel.position()).isEqualTo(20); } + @Test + public void close_afterPartialMediaRead_invokesHttpDisconnectBeforeClosingChannel() + throws IOException { + byte[] testData = new byte[100]; + Arrays.fill(testData, (byte) 0x55); + MockHttpTransport transport = mockTransport(dataRangeResponse(testData, 0, testData.length)); + + Storage storage = new Storage(transport, GsonFactory.getDefaultInstance(), r -> {}); + + GoogleCloudStorageReadOptions options = + newLazyReadOptionsBuilder().setFadvise(Fadvise.SEQUENTIAL).build(); + + HttpDisconnectCountingReadChannel readChannel = + new HttpDisconnectCountingReadChannel( + storage, + new StorageResourceId(BUCKET_NAME, OBJECT_NAME), + ApiErrorExtractor.INSTANCE, + new ClientRequestHelper<>(), + options); + + readChannel.position(0); + assertThat(readChannel.read(ByteBuffer.allocate(1))).isEqualTo(1); + + readChannel.close(); + + assertThat(readChannel.getDisconnectHttpResponseCallCount()).isEqualTo(1); + } + + @Test + public void close_withoutMediaRead_doesNotInvokeHttpDisconnect() throws IOException { + StorageObject object = newStorageObject(BUCKET_NAME, OBJECT_NAME); + MockHttpTransport transport = mockTransport(jsonDataResponse(object)); + + Storage storage = new Storage(transport, GsonFactory.getDefaultInstance(), r -> {}); + + HttpDisconnectCountingReadChannel readChannel = + new HttpDisconnectCountingReadChannel( + storage, + new StorageResourceId(BUCKET_NAME, OBJECT_NAME), + ApiErrorExtractor.INSTANCE, + new ClientRequestHelper<>(), + newLazyReadOptionsBuilder().setFastFailOnNotFoundEnabled(true).build()); + + assertThat(readChannel.size()).isEqualTo(object.getSize().longValue()); + readChannel.close(); + + assertThat(readChannel.getDisconnectHttpResponseCallCount()).isEqualTo(0); + } + + @Test + public void closeContentChannel_whenReplacingStream_invokesHttpDisconnectEachTime() + throws IOException { + byte[] testData = new byte[100]; + Arrays.fill(testData, (byte) 0x33); + MockHttpTransport transport = + mockTransport( + dataRangeResponse(testData, 0, testData.length), + dataRangeResponse(Arrays.copyOfRange(testData, 50, 100), 50, testData.length)); + + Storage storage = new Storage(transport, GsonFactory.getDefaultInstance(), r -> {}); + + GoogleCloudStorageReadOptions options = + newLazyReadOptionsBuilder().setFadvise(Fadvise.SEQUENTIAL).setInplaceSeekLimit(0).build(); + + HttpDisconnectCountingReadChannel readChannel = + new HttpDisconnectCountingReadChannel( + storage, + new StorageResourceId(BUCKET_NAME, OBJECT_NAME), + ApiErrorExtractor.INSTANCE, + new ClientRequestHelper<>(), + options); + + readChannel.position(0); + assertThat(readChannel.read(ByteBuffer.allocate(1))).isEqualTo(1); + readChannel.position(50); + assertThat(readChannel.read(ByteBuffer.allocate(1))).isEqualTo(1); + + readChannel.close(); + + assertThat(readChannel.getDisconnectHttpResponseCallCount()).isEqualTo(2); + } + + /** + * Subclass for asserting {@link GoogleCloudStorageReadChannel#disconnectHttpResponse} is invoked + * when tearing down a partially consumed media response (package-private override). + */ + private static final class HttpDisconnectCountingReadChannel + extends GoogleCloudStorageReadChannel { + + private final AtomicInteger disconnectHttpResponseCallCount = new AtomicInteger(); + + HttpDisconnectCountingReadChannel( + Storage storage, + StorageResourceId resourceId, + ApiErrorExtractor errorExtractor, + ClientRequestHelper requestHelper, + GoogleCloudStorageReadOptions readOptions) + throws IOException { + super(storage, resourceId, errorExtractor, requestHelper, readOptions); + } + + int getDisconnectHttpResponseCallCount() { + return disconnectHttpResponseCallCount.get(); + } + + @Override + void disconnectHttpResponse(HttpResponse response) { + disconnectHttpResponseCallCount.incrementAndGet(); + super.disconnectHttpResponse(response); + } + } + private static GoogleCloudStorageReadOptions.Builder newLazyReadOptionsBuilder() { return GoogleCloudStorageReadOptions.builder().setFastFailOnNotFoundEnabled(false); } From 9f5c78c4dca2d2a28876e169cb7fa688dcd77c1d Mon Sep 17 00:00:00 2001 From: Animesh Sahu Date: Tue, 9 Jun 2026 11:54:28 +0530 Subject: [PATCH 2/3] fixup --- .../hadoop/gcsio/GoogleCloudStorageReadChannel.java | 11 +++++++++-- .../gcsio/GoogleCloudStorageReadChannelTest.java | 3 ++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java index 9b9c2aac9..b227a27a9 100644 --- a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java +++ b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java @@ -556,6 +556,10 @@ protected void closeContentChannel() { * Aborts the underlying HTTP request for the current media body, if any. Must run before closing * the {@link #contentChannel} so that closing the Apache {@code ContentLengthInputStream} does * not synchronously read the rest of the response body. + * + *

Failures from {@link HttpResponse#disconnect()} are expected during normal teardown (for + * example when the socket is already closed) and are only logged here; they are not reported via + * {@link GoogleCloudStorageEventBus#postOnException()}. */ void disconnectHttpResponse(@Nullable HttpResponse response) { if (response == null) { @@ -564,7 +568,6 @@ void disconnectHttpResponse(@Nullable HttpResponse response) { try { response.disconnect(); } catch (Exception e) { - GoogleCloudStorageEventBus.postOnException(); logger.atFine().withCause(e).log( "Got an exception on HttpResponse.disconnect() for '%s'; ignoring it.", resourceId); } @@ -1168,7 +1171,11 @@ protected InputStream openStream(long bytesToRead) throws IOException { return tracked; } catch (IOException e) { GoogleCloudStorageEventBus.postOnException(); - disconnectHttpResponse(response); + try { + response.disconnect(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } throw e; } } diff --git a/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java b/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java index 07c84cfa8..efaaf3aa0 100644 --- a/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java +++ b/gcsio/src/test/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannelTest.java @@ -65,6 +65,7 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nullable; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -1168,7 +1169,7 @@ int getDisconnectHttpResponseCallCount() { } @Override - void disconnectHttpResponse(HttpResponse response) { + void disconnectHttpResponse(@Nullable HttpResponse response) { disconnectHttpResponseCallCount.incrementAndGet(); super.disconnectHttpResponse(response); } From d8ecf7570632cca4aecf287821eb62f6bee0487c Mon Sep 17 00:00:00 2001 From: Animesh Sahu Date: Tue, 16 Jun 2026 17:48:39 +0530 Subject: [PATCH 3/3] Clear openMediaResponse on natural EOF to avoid calling disconnect() on stale reference --- .../google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java | 1 + 1 file changed, 1 insertion(+) diff --git a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java index b227a27a9..75c2e4a48 100644 --- a/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java +++ b/gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageReadChannel.java @@ -372,6 +372,7 @@ public int read(ByteBuffer buffer) throws IOException { if (contentChannelEnd != size && currentPosition == contentChannelEnd) { closeContentChannel(); } else { + openMediaResponse = null; // response body fully consumed; no need to abort on close() break; } }