ratis stream read

[peterxcli]

What changes were proposed in this pull request?

Provide a one-liner summary of the changes in the PR Title field above.
It should be in the form of HDDS-1234. Short summary of the change.

Please describe your PR in detail:

• What changes are proposed in the PR? and Why? It would be better if it is written from third person's
  perspective not just for the reviewer.
• Provide as much context and rationale for the pull request as possible. It could be copy-paste from
  the Jira's description if the jira is well defined.
• If it is complex code, describe the approach used to solve the issue. If possible attach design doc,
  issue investigation, github discussion, etc.

Examples of well-written pull requests:

• #3980
• #5265
• #4701
• #5283
• #5300

What is the link to the Apache JIRA

Please create an issue in ASF JIRA before opening a pull request, and you need to set the title of the pull
request which starts with the corresponding JIRA issue number. (e.g. HDDS-XXXX. Fix a typo in YYY.)

If you do not have an ASF Jira account yet, please follow the first-time contributor
instructions in the Jira guideline.

(Please replace this section with the link to the Apache JIRA)

How was this patch tested?

(Please explain how this patch was tested. Ex: unit tests, manual tests, workflow run on the fork git repo.)
(If this patch involves UI changes, please attach a screenshot; otherwise, remove this.)

[gemini-code-assist]

Code Review

This pull request introduces the ability to read RATIS blocks through the Ratis data stream read-only API, adding RatisDataStreamBlockInputStream and associated configurations, stream observers, and benchmark tests. The review identified several critical issues in the implementation: readFully violates the standard InputStream.read contract by returning EOF instead of 0 when the target buffer is empty; a resource leak occurs in acquireClient when the acquired client is not an instance of XceiverClientRatis because it is never released; wrapping ExecutionException directly in a generic IOException hides the underlying cause; stream.write is not guaranteed to write all bytes in a single call and should be executed in a loop; and allocating a new ByteBuffer on every iteration of the loop in KeyValueHandler when rawReadBlockStream is true can cause severe GC pressure.

[gemini-code-assist]

If targetBuf has no remaining bytes, readFully will skip the loop and return EOF (-1). This violates the standard InputStream.read contract, which states that reading 0 bytes should return 0. We should return 0 immediately if targetBuf is empty.

  public synchronized int readFully(ByteBuffer targetBuf, boolean preRead)
      throws IOException {
    checkOpen();
    if (!targetBuf.hasRemaining()) {
      return 0;
    }
    int read = 0;

[gemini-code-assist]

If the acquired client is not an instance of XceiverClientRatis, an IOException is thrown, but the client is never released back to xceiverClientFactory. This causes a resource leak. Release the client before throwing the exception.

Suggested change

	if (!(client instanceof XceiverClientRatis)) {
	throw new IOException("Unexpected client class: "
	+ client.getClass().getName() + ", " + pipelineRef.get());
	}
	if (!(client instanceof XceiverClientRatis)) {
	xceiverClientFactory.releaseClientForReadData(client, false);
	throw new IOException("Unexpected client class: "
	+ client.getClass().getName() + ", " + pipelineRef.get());
	}

[gemini-code-assist]

Catching ExecutionException and wrapping it directly in a generic IOException hides the actual cause of the failure. It is better to unwrap the ExecutionException and throw its cause directly if it is an IOException.

    } catch (ExecutionException e) {
      releaseClient(true);
      final Throwable cause = e.getCause();
      if (cause instanceof IOException) {
        throw (IOException) cause;
      }
      throw new IOException("Failed Ratis read-only data stream request", cause != null ? cause : e);
    } catch (TimeoutException e) {
      releaseClient(true);
      throw new IOException("Failed Ratis read-only data stream request due to timeout", e);
    }

[gemini-code-assist]

stream.write(buffer) is not guaranteed to write all bytes in a single call, especially if the channel is non-blocking or has limited buffer space. If it writes partially, this method will throw an IOException and fail the stream read. We should write in a loop to ensure all bytes are written.

  private static void writeReadBlockStreamResponse(
      StateMachine.DataChannel stream, ContainerCommandResponseProto response,
      ByteBuffer data) throws IOException {
    final ByteBuffer buffer = encodeReadBlockStreamResponse(response, data);
    while (buffer.hasRemaining()) {
      int written = stream.write(buffer);
      if (written < 0) {
        throw new IOException("EOF reached while writing ReadBlock stream response");
      }
      if (written == 0) {
        throw new IOException("Zero bytes written to ReadBlock stream response channel");
      }
    }
  }

[gemini-code-assist]

Allocating a new ByteBuffer of responseDataSize (which can be several megabytes) on every iteration of the loop when rawReadBlockStream is true can cause severe GC pressure and performance degradation during large block reads. Consider checking if the observer actually needs to retain the buffer, or if we can safely reuse or pool it.