chungen0126 commented on code in PR #9375:
URL: https://github.com/apache/ozone/pull/9375#discussion_r2574577069
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java:
##########
@@ -207,36 +219,61 @@ protected synchronized void acquireClient() throws
IOException {
checkOpen();
if (xceiverClient == null) {
final Pipeline pipeline = pipelineRef.get();
+ final XceiverClientSpi client;
try {
- xceiverClient =
xceiverClientFactory.acquireClientForReadData(pipeline);
+ client = xceiverClientFactory.acquireClientForReadData(pipeline);
} catch (IOException ioe) {
LOG.warn("Failed to acquire client for pipeline {}, block {}",
pipeline, blockID);
throw ioe;
}
+
+ if (client == null) {
+ throw new IOException("Failed to acquire client for " + pipeline);
+ }
+ if (!(client instanceof XceiverClientGrpc)) {
+ throw new IOException("Unexpected client class: " +
client.getClass().getName() + ", " + pipeline);
+ }
+
+ xceiverClient = (XceiverClientGrpc) client;
}
}
- private void initialize() throws IOException {
- if (initialized) {
- return;
- }
- while (true) {
+ private synchronized void initialize() throws IOException {
+ while (streamingReader == null) {
try {
acquireClient();
- streamingReader = new StreamingReader();
- ContainerProtocolCalls.readBlock(xceiverClient, position, blockID,
tokenRef.get(),
- pipelineRef.get().getReplicaIndexes(), streamingReader);
- initialized = true;
- return;
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- handleExceptions(new IOException("Interrupted", ie));
} catch (IOException ioe) {
handleExceptions(ioe);
}
+
+ streamingReader = new StreamingReader();
+ xceiverClient.initStreamRead(blockID, streamingReader);
Review Comment:
Here is a simple, step-by-step explanation:
1. First read(): The code enters the initialize() method.
2. Assignment: It executes streamingReader = new StreamingReader(). The
variable is no longer null.
3. Exception: Immediately after, xceiverClient.initStreamRead() throws an
exception (e.g., connection failed).
4. Leftover State: The exception stops the process, but streamingReader is
not set back to null.
5. Second read(): The code tries to initialize() again.
6. Skipped Logic: It checks while (streamingReader == null). Since it is not
null, it skips the initialization loop.
7. Final Result: The read fails again because the stream will never
initialize.
Since the streamReader can still be reset via other methods like
`closeStream` or `seek`, I'm not sure if this behavior is functionally
acceptable.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
