chihsuan commented on code in PR #11080:
URL: https://github.com/apache/ozone/pull/11080#discussion_r3835088213
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockExtendedInputStream.java:
##########
@@ -132,7 +132,8 @@ protected void refreshBlockInfo(IOException cause, BlockID
blockID, AtomicRefere
* Check if this exception is because datanodes are not reachable.
*/
protected boolean isConnectivityIssue(IOException ex) {
- return Status.fromThrowable(ex).getCode() == Status.UNAVAILABLE.getCode();
+ final Status.Code code = Status.fromThrowable(ex).getCode();
+ return code == Status.UNAVAILABLE.getCode() || code ==
Status.DEADLINE_EXCEEDED.getCode();
Review Comment:
This also changes the classic `BlockInputStream`. `DEADLINE_EXCEEDED` now
triggers an OM block-location refresh instead of a simple retry. Is this
intentional?
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java:
##########
@@ -603,52 +611,73 @@ public void initStreamRead(BlockID blockID,
StreamingReaderSpi streamObserver) t
*/
public void initStreamRead(BlockID blockID, StreamingReaderSpi
streamObserver,
Set<DatanodeID> excludedDatanodes) throws IOException {
- final List<DatanodeDetails> datanodeList = sortDatanodes(null,
ContainerProtos.Type.ReadBlock);
- IOException lastException = null;
- for (DatanodeDetails dn : datanodeList) {
- if (excludedDatanodes.contains(dn.getID())) {
- LOG.debug("Skipping excluded datanode {} (uuid={}) for initStreamRead
{}",
- dn, dn.getUuidString(), blockID.getContainerBlockID());
- continue;
- }
- try {
- checkOpen(dn);
- semaphore.acquire();
- XceiverClientProtocolServiceStub stub =
dnChannelInfoMap.get(dn.getID()).getStub();
- if (stub == null) {
- throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
+ acquireStreamPermit(blockID);
+ boolean started = false;
+ try {
+ final List<DatanodeDetails> datanodeList = sortDatanodes(null,
ContainerProtos.Type.ReadBlock);
+ IOException lastException = null;
+ for (DatanodeDetails dn : datanodeList) {
+ if (excludedDatanodes.contains(dn.getID())) {
+ LOG.debug("Skipping excluded datanode {} (uuid={}) for
initStreamRead {}",
+ dn, dn.getUuidString(), blockID.getContainerBlockID());
+ continue;
}
- LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
- StreamObserver<ContainerCommandRequestProto> requestObserver = stub
- .withDeadlineAfter(timeout, TimeUnit.SECONDS)
- .send(streamObserver);
- streamObserver.setStreamingReadResponse(new StreamingReadResponse(dn,
- (ClientCallStreamObserver<ContainerCommandRequestProto>)
requestObserver));
- return;
- } catch (IOException e) {
- LOG.error("Failed to start streaming read to DataNode {}", dn, e);
- semaphore.release();
- lastException = e;
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted initStreamRead to " + dn + " for "
+ blockID, e);
+ try {
+ checkOpen(dn);
+ XceiverClientProtocolServiceStub stub =
dnChannelInfoMap.get(dn.getID()).getStub();
+ if (stub == null) {
+ throw new IOException("Failed to get gRPC stub for DataNode: " +
dn);
+ }
+ LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
+ // No deadline: it would bound the entire long-lived streaming call.
Per-request timeliness is
+ // enforced by streamReadTimeout in streamRead() and
StreamingReader.poll().
+ StreamObserver<ContainerCommandRequestProto> requestObserver =
stub.send(streamObserver);
Review Comment:
I may be missing an existing safeguard, but could long-lived streams keep
server-side files open for an extended period? The current limits don’t seem to
apply across the whole datanode. Is there another server-side limit or cleanup
mechanism?
--
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]
