szetszwo commented on code in PR #9375:
URL: https://github.com/apache/ozone/pull/9375#discussion_r2573196107
##########
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:
@chungen0126 , there are nothing to clean up for `StreamingReader`:
```java
//StreamingReader
private final BlockingQueue<ReadBlockResponseProto> responseQueue = new
LinkedBlockingQueue<>();
private final CompletableFuture<Void> future = new CompletableFuture<>();
private final AtomicBoolean semaphoreReleased = new AtomicBoolean(false);
private final AtomicReference<StreamingReadResponse> response = new
AtomicReference<>();
```
--
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]
