chungen0126 commented on code in PR #9375:
URL: https://github.com/apache/ozone/pull/9375#discussion_r2564922418
##########
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:
Thanks @szetszwo for the patch. Shouldn't we clean up `streamingReader`
somewhere? Otherwise, when we throw an exception here, we might never be able
to `initialize` it unless we close the entire stream.
##########
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:
Thanks @szetszwo for the patch. Should we clean up `streamingReader`
somewhere? Otherwise, when we throw an exception here, we might never be able
to `initialize` it unless we close the entire stream.
--
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]
