diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java index 90e581e8f4be..8c52f339027a 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -603,9 +604,22 @@ public void streamRead(ContainerCommandRequestProto request, @Override public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver) throws IOException { + initStreamRead(blockID, streamObserver, Collections.emptySet()); + } + + /** + * Start a streaming read, skipping datanodes that previously failed for this block stream. + */ + public void initStreamRead(BlockID blockID, StreamingReaderSpi streamObserver, + Set excludedDatanodes) throws IOException { final List 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(); diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java index dd3928f91378..975c0511832c 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java @@ -23,6 +23,8 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.time.Duration; +import java.util.HashSet; +import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; @@ -36,6 +38,8 @@ import org.apache.hadoop.fs.FSExceptionMessages; import org.apache.hadoop.hdds.StringUtils; import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto; import org.apache.hadoop.hdds.scm.OzoneClientConfig; @@ -47,6 +51,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; +import org.apache.hadoop.hdds.utils.ConnectionFailureUtils; import org.apache.hadoop.io.retry.RetryPolicy; import org.apache.hadoop.ozone.common.Checksum; import org.apache.hadoop.ozone.common.ChecksumData; @@ -91,6 +96,7 @@ public class StreamBlockInputStream extends BlockExtendedInputStream { private final Function refreshFunction; private final RetryPolicy retryPolicy; private int retries = 0; + private final Set failedStreamingDatanodes = new HashSet<>(); public StreamBlockInputStream( BlockID blockID, long length, Pipeline pipeline, @@ -171,13 +177,19 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th if (position >= blockLength) { return false; } - initialize(); - - if (bufferHasRemaining()) { - return true; + while (true) { + try { + initialize(); + if (bufferHasRemaining()) { + return true; + } + buffer = streamingReader.read(length, preRead); + retries = 0; + return bufferHasRemaining(); + } catch (IOException ex) { + handleExceptions(ex); + } } - buffer = streamingReader.read(length, preRead); - return bufferHasRemaining(); } private synchronized void advancePosition(long delta) { @@ -293,7 +305,7 @@ private synchronized void initialize() throws IOException { try { acquireClient(); final StreamingReader reader = new StreamingReader(); - xceiverClient.initStreamRead(blockID, reader); + xceiverClient.initStreamRead(blockID, reader, failedStreamingDatanodes); streamingReader = reader; } catch (IOException ioe) { handleExceptions(ioe); @@ -327,10 +339,14 @@ synchronized void readBlockImpl(long length) throws IOException { } private void handleExceptions(IOException cause) throws IOException { - if (cause instanceof StorageContainerException || isConnectivityIssue(cause)) { - if (shouldRetryRead(cause, retryPolicy, retries++)) { + IOException root = ConnectionFailureUtils.unwrapCause(cause); + if (root instanceof StorageContainerException || isConnectivityIssue(root) || + root instanceof TimeoutIOException) { + if (shouldRetryRead(root, retryPolicy, retries++)) { + recordFailedStreamingDatanode(); releaseClient(); - refreshBlockInfo(cause); + refreshBlockInfo(root); + requestedLength = position; LOG.warn("Refreshing block data to read block {} due to {}", blockID, cause.getMessage()); } else { throw cause; @@ -340,6 +356,21 @@ private void handleExceptions(IOException cause) throws IOException { } } + private void recordFailedStreamingDatanode() { + if (streamingReader == null) { + return; + } + final StreamingReadResponse response = streamingReader.getResponse(); + if (response == null) { + return; + } + final DatanodeDetails dn = response.getDatanodeDetails(); + if (failedStreamingDatanodes.add(dn.getID())) { + LOG.warn("Excluding DataNode {} from streaming read retries for block {}", + dn, blockID); + } + } + protected synchronized void releaseClient() { if (xceiverClientFactory != null && xceiverClient != null) { closeStream(); diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java index 4cdf13032b47..9e1ff0893ecc 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java @@ -224,7 +224,7 @@ public void testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws Excep reader.setStreamingReadResponse(streamingReadResponse); readerRef.set(reader); return null; - }).when(xceiverClient).initStreamRead(any(BlockID.class), any()); + }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any()); // Simulate the race: when the client sends a ReadBlock request, the server // responds with data (onNext) and closes the stream (onCompleted) before @@ -304,7 +304,7 @@ private XceiverClientGrpc mockStreamingReadClient(byte[] data, .setReadBlock(readBlock) .build()); return null; - }).when(xceiverClient).initStreamRead(any(BlockID.class), any()); + }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any()); return xceiverClient; } @@ -338,7 +338,7 @@ public void testReadDoesNotDropQueuedItemsWhenFutureIsDoneOnSecondCall() throws reader.setStreamingReadResponse(streamingReadResponse); readerRef.set(reader); return null; - }).when(xceiverClient).initStreamRead(any(BlockID.class), any()); + }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any()); // Server delivers both 4-byte chunks plus onCompleted() in one synchronous // call. After streamRead() returns: queue=[chunk1, chunk2], isDone=true. @@ -388,7 +388,7 @@ public void testReadGetsFreshResponseTimeoutAfterStreamReadWait() throws Excepti reader.setStreamingReadResponse(streamingReadResponse); readerRef.set(reader); return null; - }).when(xceiverClient).initStreamRead(any(BlockID.class), any()); + }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any()); doAnswer(inv -> { Thread.sleep(450); Thread responseThread = new Thread(() -> { @@ -437,7 +437,7 @@ public void testReadWithoutNewRequestGetsFreshTimeoutBudget() throws Exception { reader.setStreamingReadResponse(streamingReadResponse); readerRef.set(reader); return null; - }).when(xceiverClient).initStreamRead(any(BlockID.class), any()); + }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any()); doAnswer(inv -> { streamReads.incrementAndGet(); readerRef.get().onNext(buildResponseProto(new byte[] {1}, 0)); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java index 62cd868c763a..d7f45b1cd7e3 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java @@ -18,11 +18,15 @@ package org.apache.hadoop.hdds.utils; import java.io.EOFException; +import java.io.IOException; import java.net.ConnectException; import java.net.NoRouteToHostException; import java.net.SocketException; import java.net.SocketTimeoutException; import java.net.UnknownHostException; +import java.util.concurrent.ExecutionException; +import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; +import org.apache.ratis.protocol.exceptions.TimeoutIOException; /** * Shared classifier for exceptions where the cached peer IP is no longer @@ -100,4 +104,28 @@ public static boolean isConnectionFailure(Throwable t) { } return false; } + + /** + * Returns the first {@link StorageContainerException} or + * {@link TimeoutIOException} in {@code ex}'s cause chain (through + * {@link ExecutionException} and nested {@link IOException} wrappers). + */ + public static IOException unwrapCause(IOException ex) { + Throwable t = ex; + while (t != null) { + if (t instanceof TimeoutIOException || t instanceof StorageContainerException) { + return (IOException) t; + } + if (t instanceof ExecutionException && t.getCause() != null) { + t = t.getCause(); + continue; + } + if (t.getCause() instanceof IOException) { + t = t.getCause(); + continue; + } + break; + } + return ex; + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java new file mode 100644 index 000000000000..27a971873be6 --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java @@ -0,0 +1,287 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.client.rpc.read; + +import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE; +import static org.apache.hadoop.ozone.client.OzoneClientTestUtils.assertKeyContent; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.apache.commons.lang3.RandomUtils; +import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.hdds.scm.StreamingReadResponse; +import org.apache.hadoop.hdds.scm.container.ContainerID; +import org.apache.hadoop.hdds.scm.container.ContainerInfo; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.server.StorageContainerManager; +import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream; +import org.apache.hadoop.hdds.utils.IOUtils; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.TestDataUtil; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneClientFactory; +import org.apache.hadoop.ozone.client.io.KeyInputStream; +import org.apache.hadoop.ozone.om.TestBucket; +import org.apache.hadoop.ozone.om.helpers.OmKeyArgs; +import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +/** + * Verifies that streaming block reads fail over to a healthy replica when the + * datanode serving the stream becomes unavailable, matching legacy + * {@code BlockInputStream} behavior. + * + *

With {@code ozone.client.stream.readblock.enable=true}, reads currently + * do not fail over correctly when the streaming datanode stops responding. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class TestStreamReadDatanodeFailover { + + private static final Duration STREAM_READ_TIMEOUT = Duration.ofSeconds(3); + private static final int PIPELINE_READY_TIMEOUT_MS = 30_000; + + private MiniOzoneCluster cluster; + private final Set stoppedDatanodes = new HashSet<>(); + + @BeforeAll + void setup() throws Exception { + cluster = newCluster(); + cluster.waitForClusterToBeReady(); + } + + @BeforeEach + void waitForPipeline() throws Exception { + cluster.waitForPipelineTobeReady(THREE, PIPELINE_READY_TIMEOUT_MS); + } + + @AfterEach + void resetDatanodes() throws Exception { + if (stoppedDatanodes.isEmpty()) { + return; + } + List toRestart = new ArrayList<>(stoppedDatanodes); + stoppedDatanodes.clear(); + for (DatanodeDetails dn : toRestart) { + cluster.restartHddsDatanode(dn, false); + } + cluster.waitForClusterToBeReady(); + } + + @AfterAll + void cleanup() { + IOUtils.closeQuietly(cluster); + } + + /** + * Legacy (Async) chunk reads succeed after stopping one pipeline datanode. + */ + @Test + void testAsyncReadFailoverWhenDatanodeStopped() throws Exception { + try (OzoneClient client = cluster.newClient()) { + ObjectStore store = client.getObjectStore(); + OzoneBucket bucket = createBucket(store); + String keyName = newKeyName(); + byte[] content = RandomUtils.secure().randomBytes(32 * 1024); + TestDataUtil.createKey(bucket, keyName, + RatisReplicationConfig.getInstance(THREE), content); + + List datanodes = getPipelineDatanodes(cluster, bucket, keyName); + assertEquals(3, datanodes.size()); + + stopDatanode(datanodes.get(0)); + assertKeyContent(bucket, keyName, content); + + stopDatanode(datanodes.get(1)); + assertKeyContent(bucket, keyName, content); + } + } + + /** + * Streaming reads should succeed after stopping pipeline datanodes when at + * least one healthy replica remains, same as the legacy path. + */ + @Test + void testStreamReadFailoverWhenDatanodesStopped() throws Exception { + OzoneConfiguration streamConf = streamReadConfig(cluster.getConf()); + try (OzoneClient client = OzoneClientFactory.getRpcClient(streamConf)) { + ObjectStore store = client.getObjectStore(); + OzoneBucket bucket = createBucket(store); + String keyName = newKeyName(); + byte[] content = RandomUtils.secure().randomBytes(32 * 1024); + TestDataUtil.createKey(bucket, keyName, + RatisReplicationConfig.getInstance(THREE), content); + + List datanodes = getPipelineDatanodes(cluster, bucket, keyName); + assertEquals(3, datanodes.size()); + + // Sanity check: streaming read works with all datanodes up. + assertKeyContent(bucket, keyName, content); + + stopDatanode(datanodes.get(0)); + assertKeyContent(bucket, keyName, content); + + stopDatanode(datanodes.get(1)); + assertKeyContent(bucket, keyName, content); + } + } + + /** + * After a streaming read has started, stopping the datanode serving the + * stream must not break the read. Legacy reads fail over and continue from + * the current offset; streaming reads currently fail (timeout or wrong data). + */ + @Test + void testStreamReadFailoverAfterActiveDatanodeStopped() throws Exception { + OzoneConfiguration streamConf = streamReadConfig(cluster.getConf()); + try (OzoneClient streamClient = OzoneClientFactory.getRpcClient(streamConf); + OzoneClient legacyClient = cluster.newClient()) { + TestBucket streamBucket = TestBucket.newBuilder(streamClient).build(); + OzoneBucket legacyBucket = legacyClient.getObjectStore() + .getVolume(streamBucket.delegate().getVolumeName()) + .getBucket(streamBucket.delegate().getName()); + + String keyName = newKeyName(); + byte[] content = streamBucket.writeRandomBytes(keyName, 32 * 1024); + + List datanodes = + getPipelineDatanodes(cluster, streamBucket.delegate(), keyName); + assertEquals(3, datanodes.size()); + + // Legacy control: stopping one pipeline datanode mid-read still succeeds. + try (InputStream legacyIn = legacyBucket.readKey(keyName)) { + assertNotEquals(-1, legacyIn.read()); + stopDatanode(datanodes.get(0)); + readRemaining(legacyIn, content, 1); + } + + // Streaming read: stop the datanode actively serving the stream. + // This fails today without a proper streaming read failover fix. + try (KeyInputStream streamIn = streamBucket.getKeyInputStream(keyName)) { + assertNotEquals(-1, streamIn.read()); + StreamBlockInputStream blockStream = + (StreamBlockInputStream) streamIn.getPartStreams().get(0); + DatanodeDetails activeDatanode = getActiveStreamingDatanode(blockStream); + assertTrue(datanodes.contains(activeDatanode), + "Active streaming datanode should belong to the key pipeline"); + stopDatanode(activeDatanode); + readRemaining(streamIn, content, 1); + } + } + } + + private void stopDatanode(DatanodeDetails dn) throws IOException { + cluster.shutdownHddsDatanode(dn); + stoppedDatanodes.add(dn); + } + + private static void readRemaining(InputStream in, byte[] expected, int offset) + throws IOException { + byte[] actual = org.apache.commons.io.IOUtils.readFully(in, expected.length - offset); + assertArrayEquals( + java.util.Arrays.copyOfRange(expected, offset, expected.length), + actual); + } + + private static DatanodeDetails getActiveStreamingDatanode(StreamBlockInputStream blockStream) + throws ReflectiveOperationException { + Field streamingReaderField = StreamBlockInputStream.class.getDeclaredField("streamingReader"); + streamingReaderField.setAccessible(true); + Object streamingReader = streamingReaderField.get(blockStream); + assertTrue(streamingReader != null, "Streaming reader should be initialized after first read"); + + Method getResponse = streamingReader.getClass().getDeclaredMethod("getResponse"); + getResponse.setAccessible(true); + StreamingReadResponse response = (StreamingReadResponse) getResponse.invoke(streamingReader); + assertTrue(response != null, "Streaming read response should be initialized"); + return response.getDatanodeDetails(); + } + + private static OzoneConfiguration streamReadConfig(OzoneConfiguration base) { + OzoneClientConfig clientConfig = base.getObject(OzoneClientConfig.class); + clientConfig.setStreamReadBlock(true); + clientConfig.setStreamReadTimeout(STREAM_READ_TIMEOUT); + OzoneConfiguration conf = new OzoneConfiguration(base); + conf.setFromObject(clientConfig); + return conf; + } + + private static MiniOzoneCluster newCluster() throws IOException { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT, 1); + conf.setInt(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT, 1); + conf.setInt(ScmConfigKeys.OZONE_SCM_RATIS_PIPELINE_LIMIT, 5); + return MiniOzoneCluster.newBuilder(conf) + .setNumDatanodes(3) + .build(); + } + + private static OzoneBucket createBucket(ObjectStore store) throws IOException { + String volumeName = UUID.randomUUID().toString(); + store.createVolume(volumeName); + String bucketName = UUID.randomUUID().toString(); + store.getVolume(volumeName).createBucket(bucketName); + return store.getVolume(volumeName).getBucket(bucketName); + } + + private static String newKeyName() { + return "key-" + UUID.randomUUID(); + } + + private static List getPipelineDatanodes(MiniOzoneCluster cluster, + OzoneBucket bucket, String keyName) throws IOException { + OmKeyArgs keyArgs = new OmKeyArgs.Builder() + .setVolumeName(bucket.getVolumeName()) + .setBucketName(bucket.getName()) + .setKeyName(keyName) + .build(); + OmKeyLocationInfo keyInfo = cluster.getOzoneManager().lookupKey(keyArgs) + .getKeyLocationVersions().get(0) + .getBlocksLatestVersionOnly().get(0); + long containerID = keyInfo.getContainerID(); + + StorageContainerManager scm = cluster.getStorageContainerManager(); + ContainerInfo container = scm.getContainerManager() + .getContainer(ContainerID.valueOf(containerID)); + Pipeline pipeline = scm.getPipelineManager() + .getPipeline(container.getPipelineID()); + return pipeline.getNodes(); + } +}