ResponseStream.close() can block on the reader lock before closing the HTTP response

src/main/java/com/google/genai/ResponseStream.javaから始め、特にcloseOwnResponse()とreader.readLine()の反復処理経路を確認してください。HttpApiResponse.close()がOkHttpレスポンスをどのように閉じるかも確認してください。提供された回帰テストをResponseStreamTestと併せて実行し、readerロックの待機を再現してください。レスポンスを閉じるとreaderのクリーンアップより先にブロックされた読み取りが解放され、両方のテストが成功すれば完了です。

Description

This is a client library issue in ResponseStream, rather than a Gemini service issue.

ResponseStream.close() currently closes the BufferedReader before closing the HTTP response. If another thread is blocked in reader.readLine(), closing the reader can wait for the lock held by that read. As a result, execution cannot reach response.close(), which could otherwise release the underlying read.

This can delay stream cleanup until the read completes, fails, or times out.

Environment details

• Programming language: Java
• OS: macOS 27.0.1 (arm64)
• Language runtime version: Eclipse Temurin OpenJDK 25.0.4+7-LTS
• Package version: com.google.genai:google-genai:1.81.0-SNAPSHOT (source checkout)
• Commit: 79d09a6cee17c0c816fb2c7014ad5a92770a9d4c
• OkHttp version: 4.12.0

Relevant code

ResponseStream.closeOwnResponse() currently executes:

try {
  // ...
  reader.close();
  // ...
} finally {
  // ...
  response.close();
}

The reader is a BufferedReader wrapping an InputStreamReader over the response body, and stream iteration calls reader.readLine(). HttpApiResponse.close() delegates to the OkHttp response's close().

The finally block does not prevent this wait: it is only entered after reader.close() returns or throws.

Steps to reproduce

1. Check out googleapis/java-genai at commit 79d09a6cee17c0c816fb2c7014ad5a92770a9d4c (1.81.0-SNAPSHOT).
2. Save the test below as src/test/java/com/google/genai/ResponseStreamCloseTest.java.
3. Run mvn test -Dtest=ResponseStreamCloseTest -Djacoco.skip=true from the repository root.

The test uses the SDK's actual HttpApiResponse, an OkHttp response/body, and a controlled Okio Source. No HTTP server, credentials, or additional dependencies are required.

The source signals when a read is in progress and waits until it is closed. The test then closes the stream on another thread and checks that the body is closed and both operations finish. Its finally block releases the read even on failure, so the current implementation does not hang the test suite.

package com.google.genai;

import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.genai.types.GenerateContentResponse;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import okhttp3.MediaType;
import okhttp3.Protocol;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.Buffer;
import okio.Okio;
import okio.Source;
import okio.Timeout;
import org.junit.jupiter.api.Test;

public final class ResponseStreamCloseTest {

  @Test
  public void testCloseUnblocksReadInProgress() throws Exception {
    BlockingSource source = new BlockingSource();
    ResponseBody body =
        ResponseBody.create(MediaType.parse("text/event-stream"), -1, Okio.buffer(source));
    // Use the SDK's real response wrapper and OkHttp's real close path; no network is needed.
    HttpApiResponse response =
        new HttpApiResponse(
            new Response.Builder()
                .request(new Request.Builder().url("https://example.com/stream").build())
                .protocol(Protocol.HTTP_1_1)
                .code(200)
                .message("OK")
                .body(body)
                .build());
    ResponseStream<GenerateContentResponse> stream =
        new ResponseStream<>(
            GenerateContentResponse.class, response, new DummyConverter(), "convert");
    ExecutorService executor =
        Executors.newFixedThreadPool(
            2,
            runnable -> {
              Thread thread = new Thread(runnable);
              thread.setDaemon(true);
              return thread;
            });

    try {
      Future<Boolean> reading = executor.submit(() -> stream.iterator().hasNext());
      // Once the source is reading, BufferedReader.readLine() already holds the reader's lock.
      assertTrue(source.readStarted.await(5, TimeUnit.SECONDS), "Stream read did not start");

      Future<?> closing = executor.submit(() -> stream.close());

      // Closing the response body must release the read before reader.close() waits for its lock.
      assertTrue(
          source.closed.await(5, TimeUnit.SECONDS),
          "Response body was not closed while a read held the reader lock;"
              + " close the HTTP response before closing the reader");
      closing.get(5, TimeUnit.SECONDS);
      assertFalse(reading.get(5, TimeUnit.SECONDS), "Closing the body should end the stream");
    } finally {
      // Also release the read on failure, so the broken implementation cannot hang the test suite.
      source.releaseRead.countDown();
      executor.shutdown();
      assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "Stream workers did not finish");
      stream.close();
    }
  }

  public static final class DummyConverter {
    public JsonNode convert(JsonNode fromObject, ObjectNode parentObject) {
      return fromObject;
    }
  }

  private static final class BlockingSource implements Source {
    private final CountDownLatch readStarted = new CountDownLatch(1);
    private final CountDownLatch releaseRead = new CountDownLatch(1);
    private final CountDownLatch closed = new CountDownLatch(1);

    @Override
    public long read(Buffer sink, long byteCount) throws IOException {
      readStarted.countDown();
      try {
        releaseRead.await();
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new IOException("Interrupted while waiting for the response body to close", e);
      }
      return -1;
    }

    @Override
    public Timeout timeout() {
      return Timeout.NONE;
    }

    @Override
    public void close() {
      closed.countDown();
      releaseRead.countDown();
    }
  }
}

Actual behavior

With the current implementation, the test fails at the body-closure assertion after its five-second wait:

Tests run: 1, Failures: 1, Errors: 0, Skipped: 0
Response body was not closed while a read held the reader lock;
close the HTTP response before closing the reader

In a temporary copy of the checkout, reversing only the close order makes this exact test pass. Running it together with the existing ResponseStreamTest also passes:

Tests run: 6, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

Expected behavior / suggested change

Close the HTTP response before attempting to acquire the reader's lock, and close the reader in a finally block so its cleanup is still attempted if closing the response throws. Preserve the existing IOException handling for the reader.

The regression test above verifies this behavior by starting a blocking read on one thread and closing the stream on another, without manually releasing the read before the assertions.

Scope

This regression test demonstrates the reader-lock dependency and the effect of reversing the close order when response closure releases the read. It does not measure cancellation latency against a real server or establish that response closure immediately interrupts every HTTP transport. Cancellation before a response is available is a separate concern.

Related reports / duplicate check (2026-10-29)

I searched open and closed issues and pull requests in googleapis/java-genai using ResponseStream, close, cancel, deadlock, leak, BufferedReader, stream, hang, and block, and reviewed the current open PR list. I did not find a direct duplicate describing reader.close() waiting for the reader lock before HTTP response closure.

Related reports have different scopes:

• java-genai #517 reports malformed/truncated SSE parsing and connection-leak warnings. It was closed for inactivity; its discussion does not identify this close-order lock dependency.
• java-genai PR #951 was merged to handle SSE error messages and multiline payloads. Its ResponseStream.java diff does not change the close order.
• java-genai #183 concerns concurrent non-streaming calls. The reporter later attributed it to virtual-thread pinning, rather than stream closure.

The upstream main HEAD was also 79d09a6cee17c0c816fb2c7014ad5a92770a9d4c at the time of this check, matching the checkout used for reproduction. These search results do not establish the absence of an unindexed or differently described report.

Contribution

Would you be open to a PR addressing the close-order issue in ResponseStream.close()?

The proposed fix closes the HTTP response first, then closes the reader in a finally block. I have a regression test that fails with the current implementation and passes with this change. The existing ResponseStreamTest tests also pass.

I noticed that CONTRIBUTING.md says contributions will be accepted in the future, so I wanted to check before opening a PR. If contributions are welcome, I'd be happy to submit the fix and regression test.