From c688981cff08c62472006f995ccefd04856a1bdb Mon Sep 17 00:00:00 2001 From: peterxcli Date: Tue, 6 Oct 2026 00:33:49 +0800 Subject: [PATCH 1/7] HDDS-16722. Serve ReadBlock over Ratis read-only data streams on the datanode --- .../apache/hadoop/hdds/DatanodeVersion.java | 3 + .../ratis/ClosedContainerReadResolver.java | 69 ++++++++ .../server/ratis/ContainerStateMachine.java | 105 ++++++++++++ .../server/ratis/XceiverServerRatis.java | 5 + .../ratis/ContainerStateMachineTests.java | 158 ++++++++++++++++++ .../TestClosedContainerReadResolver.java | 111 ++++++++++++ .../ozone/client/rpc/read/TestStreamRead.java | 100 +++++++++++ 7 files changed, 551 insertions(+) create mode 100644 hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java create mode 100644 hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java index 560a1edf123b..f887a758065e 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java @@ -38,6 +38,9 @@ public enum DatanodeVersion implements ComponentVersion { SHORT_CIRCUIT_READS(4, "Version with short-circuit read support."), + RATIS_DATASTREAM_READ_BLOCK_SUPPORT(5, "This version supports ReadBlock on Ratis read-only data streams, " + + "also for closed containers without a Raft group."), + FUTURE_VERSION(-1, "Used internally in the client when the server side is " + " newer and an unknown server version has arrived to the client."); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java new file mode 100644 index 000000000000..6692305d672d --- /dev/null +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java @@ -0,0 +1,69 @@ +/* + * 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.container.common.transport.server.ratis; + +import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED; + +import java.io.IOException; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; +import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; +import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.server.api.DataStreamApi; + +/** + * Serves ReadBlock on a Ratis read-only data stream for a closed container, without a Raft group. + * SCM gives the read pipeline of a closed container a random ID, and no Raft group has this ID. + * A closed container does not change, so this datanode can serve the read from its replica. + */ +final class ClosedContainerReadResolver implements DataStreamApi.Resolver { + private final ContainerDispatcher dispatcher; + private final ContainerController containerController; + private final String datanodeUuid; + + ClosedContainerReadResolver(ContainerDispatcher dispatcher, ContainerController containerController, + DatanodeDetails datanode) { + this.dispatcher = dispatcher; + this.containerController = containerController; + this.datanodeUuid = datanode.getUuidString(); + } + + /** + * @return the API to serve a ReadBlock of a closed container on this datanode. Otherwise, null: Ratis then serves + * the request with its Raft group. + */ + @Override + public DataStreamApi resolve(RaftClientRequest request) throws IOException { + final ContainerCommandRequestProto requestProto = ContainerCommandRequestMessage.toProto( + request.getMessage().getContent(), request.getRaftGroupId()); + if (requestProto.getCmdType() != Type.ReadBlock + || requestProto.hasDatanodeUuid() && !datanodeUuid.equals(requestProto.getDatanodeUuid())) { + return null; + } + + final Container container = containerController.getContainer(requestProto.getContainerID()); + if (container == null || container.getContainerState() != CLOSED) { + return null; + } + return (message, stream) -> ContainerStateMachine.streamReadBlock(dispatcher, requestProto, stream); + } +} diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index 9a3e988e27ad..4c423039bdcb 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -28,6 +28,9 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; import java.nio.file.Files; import java.util.Arrays; import java.util.Collection; @@ -52,8 +55,11 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.hadoop.hdds.HddsUtils; +import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.StorageUnit; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -70,6 +76,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.hdds.utils.Cache; import org.apache.hadoop.hdds.utils.ResourceCache; +import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel; import org.apache.hadoop.ozone.HddsDatanodeService; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.common.utils.BufferUtils; @@ -106,6 +113,7 @@ import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException; import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat; +import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.apache.ratis.util.FileUtils; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.LifeCycle; @@ -121,6 +129,7 @@ *

* The container requests can be divided into readonly request, WriteChunk request and other write requests. * - Read only requests (see {@link HddsUtils#isReadOnly}) are handled by {@link #query(Message)}. + * - ReadBlock requests on a Ratis read-only data stream are handled by {@link #transferTo}. * - WriteChunk request contains user data * - Other write request does not contain user data. *

@@ -860,6 +869,102 @@ public CompletableFuture query(Message request) { } } + /** Serves a Ratis read-only data stream. Only ReadBlock is supported, see {@link #streamReadBlock}. */ + @Override + public long transferTo(Message request, WritableByteChannel stream) throws IOException { + try { + metrics.incNumQueryStateMachineOps(); + final ContainerCommandRequestProto requestProto = message2ContainerCommandRequestProto(request); + if (requestProto.getCmdType() != Type.ReadBlock) { + throw new IOException("Unsupported command for a read-only data stream: " + requestProto.getCmdType()); + } + return streamReadBlock(dispatcher, requestProto, stream); + } catch (IOException e) { + metrics.incNumQueryStateMachineFails(); + throw e; + } + } + + /** + * Reads a block and writes each ReadBlock response to the stream as one reply: + * {@code [int: metadata length][metadata: the response without its data][data]}. + * Then closes the stream, which sends the terminal reply. + * If the read fails, this method throws without closing the stream. Ratis sends the failure as the terminal reply. + * + * @return the number of bytes written to the stream + */ + static long streamReadBlock(ContainerDispatcher dispatcher, ContainerCommandRequestProto request, + WritableByteChannel stream) throws IOException { + final AtomicLong bytesWritten = new AtomicLong(); + final AtomicReference error = new AtomicReference<>(); + final StreamObserver observer = new StreamObserver() { + @Override + public void onNext(ContainerCommandResponseProto response) { + if (error.get() != null) { + // The read already failed. Drop the error response that the dispatcher sends after a failed write. + return; + } + try { + bytesWritten.addAndGet(writeReadBlockReply(stream, response)); + } catch (IOException e) { + error.set(e); + // Throw to stop reading the block + throw new UncheckedIOException(e); + } + } + + @Override + public void onError(Throwable t) { + error.compareAndSet(null, t); + } + + @Override + public void onCompleted() { + } + }; + + try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) { + dispatcher.streamDataReadOnly(request, observer, blockFile, null); + } catch (UncheckedIOException e) { + error.compareAndSet(null, e.getCause()); + } + if (error.get() != null) { + throw new IOException("Failed to read block " + BlockID.getFromProtobuf(request.getReadBlock().getBlockID()), + error.get()); + } + stream.close(); + return bytesWritten.get(); + } + + private static int writeReadBlockReply(WritableByteChannel stream, ContainerCommandResponseProto response) + throws IOException { + final ByteString data; + final ContainerCommandResponseProto metadata; + if (response.hasReadBlock()) { + data = response.getReadBlock().getData(); + metadata = response.toBuilder() + .setReadBlock(response.getReadBlock().toBuilder().setData(ByteString.EMPTY)) + .build(); + } else { + data = ByteString.EMPTY; + metadata = response; + } + final ByteString metadataBytes = metadata.toByteString(); + final ByteBuffer reply = ByteBuffer.allocate(Integer.BYTES + metadataBytes.size() + data.size()); + reply.putInt(metadataBytes.size()); + metadataBytes.copyTo(reply); + data.copyTo(reply); + reply.flip(); + + // The client parses each reply on its own, so the reply must be sent in one write. + final int length = reply.remaining(); + final int written = stream.write(reply); + if (written != length) { + throw new IOException("Failed to write a ReadBlock reply: only " + written + " of " + length + " bytes written"); + } + return length; + } + private ByteString readStateMachineData( ContainerCommandRequestProto requestProto, long term, long index) throws IOException { diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java index 90c413088e8d..2ea583a6e510 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java @@ -526,6 +526,11 @@ public static XceiverServerRatis newXceiverServerRatis(HddsDatanodeService hddsD CertificateClient caClient, StateContext context) throws IOException { Parameters parameters = createTlsParameters( new SecurityConfig(ozoneConf), caClient); + if (parameters == null) { + parameters = new Parameters(); + } + RaftServerConfigKeys.DataStream.setServerApiResolver(parameters, + new ClosedContainerReadResolver(dispatcher, containerController, datanodeDetails)); return new XceiverServerRatis(hddsDatanodeService, datanodeDetails, dispatcher, containerController, context, ozoneConf, parameters); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java index 8ad92ec358e9..ae2ddf954e34 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java @@ -19,12 +19,16 @@ import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_CONTAINER_RATIS_STATEMACHINE_WRITE_WAIT_INTERVAL; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -32,6 +36,11 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -40,11 +49,13 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; +import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; @@ -58,6 +69,9 @@ import org.apache.ratis.server.RaftServer; import org.apache.ratis.statemachine.TransactionContext; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; +import org.apache.ratis.thirdparty.io.grpc.Status; +import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; +import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -247,6 +261,150 @@ public void testWriteTimout() throws Exception { assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); } + @Test + public void testTransferToReadBlock() throws Exception { + final ByteString first = ByteString.copyFromUtf8("first data"); + final List responses = Arrays.asList( + readBlockResponse(0, first), readBlockResponse(first.size(), ByteString.copyFromUtf8("second data"))); + setUpMockReadBlock(observer -> responses.forEach(observer::onNext)); + final List replies = new ArrayList<>(); + final WritableByteChannel stream = newReadStream(replies); + + final long written = stateMachine.transferTo(newRequest(ContainerProtos.Type.ReadBlock), stream); + + // Each response is one reply: [metadata length][the response without its data][data] + assertEquals(responses.size(), replies.size()); + long replyBytes = 0; + for (int i = 0; i < responses.size(); i++) { + final ContainerProtos.ContainerCommandResponseProto response = responses.get(i); + final ByteBuffer reply = replies.get(i); + replyBytes += reply.remaining(); + assertEquals(readBlockResponse(response.getReadBlock().getOffset(), ByteString.EMPTY), parseMetadata(reply)); + assertEquals(response.getReadBlock().getData(), ByteString.copyFrom(reply)); + } + assertEquals(replyBytes, written); + verify(stream).close(); + } + + @Test + public void testTransferToReadBlockErrorResponse() throws Exception { + final ContainerProtos.ContainerCommandResponseProto error = errorResponse(); + setUpMockReadBlock(observer -> observer.onNext(error)); + final List replies = new ArrayList<>(); + final WritableByteChannel stream = newReadStream(replies); + + stateMachine.transferTo(newRequest(ContainerProtos.Type.ReadBlock), stream); + + assertEquals(1, replies.size()); + assertEquals(error, parseMetadata(replies.get(0))); + assertFalse(replies.get(0).hasRemaining()); + verify(stream).close(); + } + + /** The stream stays open, so Ratis sends the failure as the terminal reply. */ + @Test + public void testTransferToReadBlockReadFailure() throws Exception { + final StatusRuntimeException failure = Status.OUT_OF_RANGE.asRuntimeException(); + setUpMockReadBlock(observer -> observer.onError(failure)); + final WritableByteChannel stream = mock(WritableByteChannel.class); + + final IOException e = assertThrows(IOException.class, + () -> stateMachine.transferTo(newRequest(ContainerProtos.Type.ReadBlock), stream)); + + assertSame(failure, e.getCause()); + verify(stream, never()).write(any()); + verify(stream, never()).close(); + } + + @Test + public void testTransferToReadBlockWriteFailure() throws Exception { + setUpMockReadBlock(observer -> { + // The failed write stops the read. Then the dispatcher sends the error response of the read. + assertThrows(UncheckedIOException.class, () -> observer.onNext(readBlockResponse(0, ByteString.EMPTY))); + observer.onNext(errorResponse()); + }); + final IOException failure = new IOException("Connection closed"); + final WritableByteChannel stream = mock(WritableByteChannel.class); + when(stream.write(any())).thenThrow(failure); + + final IOException e = assertThrows(IOException.class, + () -> stateMachine.transferTo(newRequest(ContainerProtos.Type.ReadBlock), stream)); + + assertSame(failure, e.getCause()); + // The error response is not written to the failed stream. + verify(stream, times(1)).write(any()); + verify(stream, never()).close(); + } + + @Test + public void testTransferToUnsupportedCommand() throws Exception { + final WritableByteChannel stream = mock(WritableByteChannel.class); + + assertThrows(IOException.class, () -> stateMachine.transferTo(newRequest(ContainerProtos.Type.GetBlock), stream)); + + verify(dispatcher, never()).streamDataReadOnly(any(), any(), any(), any()); + verify(stream, never()).write(any()); + } + + private void setUpMockReadBlock(Consumer> readBlock) { + doAnswer(invocation -> { + readBlock.accept(invocation.getArgument(1)); + return null; + }).when(dispatcher).streamDataReadOnly(any(), any(), any(), any()); + } + + private static Message newRequest(ContainerProtos.Type type) { + return ContainerCommandRequestMessage.toMessage(ContainerProtos.ContainerCommandRequestProto.newBuilder() + .setCmdType(type) + .setContainerID(1) + .setDatanodeUuid(UUID.randomUUID().toString()) + .build(), null); + } + + private static ContainerProtos.ContainerCommandResponseProto readBlockResponse(long offset, ByteString data) { + return ContainerProtos.ContainerCommandResponseProto.newBuilder() + .setCmdType(ContainerProtos.Type.ReadBlock) + .setResult(ContainerProtos.Result.SUCCESS) + .setReadBlock(ContainerProtos.ReadBlockResponseProto.newBuilder() + .setChecksumData(ContainerProtos.ChecksumData.newBuilder() + .setType(ContainerProtos.ChecksumType.NONE) + .setBytesPerChecksum(1024)) + .setOffset(offset) + .setData(data)) + .build(); + } + + private static ContainerProtos.ContainerCommandResponseProto errorResponse() { + return ContainerProtos.ContainerCommandResponseProto.newBuilder() + .setCmdType(ContainerProtos.Type.ReadBlock) + .setResult(ContainerProtos.Result.CONTAINER_NOT_FOUND) + .setMessage("Container not found") + .build(); + } + + /** @return a mock Ratis read-only stream, which sends each buffer as one reply and does not move its position. */ + private static WritableByteChannel newReadStream(List replies) throws IOException { + final WritableByteChannel stream = mock(WritableByteChannel.class); + when(stream.write(any())).thenAnswer(invocation -> { + final ByteBuffer buffer = invocation.getArgument(0); + final ByteBuffer reply = ByteBuffer.allocate(buffer.remaining()); + reply.put(buffer.duplicate()); + reply.flip(); + replies.add(reply); + return reply.remaining(); + }); + return stream; + } + + /** Parses the metadata of a reply, and moves the reply to its data. */ + private static ContainerProtos.ContainerCommandResponseProto parseMetadata(ByteBuffer reply) throws IOException { + final int length = reply.getInt(); + final ByteBuffer metadata = reply.slice(); + metadata.limit(length); + reply.position(reply.position() + length); + return ContainerProtos.ContainerCommandResponseProto.parseFrom(metadata); + } + private void setUpMockDispatcherReturn(boolean failWithException) { if (failWithException) { when(dispatcher.dispatch(any(), any())).thenThrow(new RuntimeException()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java new file mode 100644 index 000000000000..53d4a805197e --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java @@ -0,0 +1,111 @@ +/* + * 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.container.common.transport.server.ratis; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.UUID; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; +import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; +import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.RaftGroupId; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +/** + * Tests {@link ClosedContainerReadResolver}. + */ +class TestClosedContainerReadResolver { + private static final long CONTAINER_ID = 1L; + + private ContainerController containerController; + private Container container; + private ClosedContainerReadResolver resolver; + private String datanodeUuid; + + @BeforeEach + void setUp() { + datanodeUuid = UUID.randomUUID().toString(); + final DatanodeDetails datanode = mock(DatanodeDetails.class); + when(datanode.getUuidString()).thenReturn(datanodeUuid); + containerController = mock(ContainerController.class); + container = mock(Container.class); + when(containerController.getContainer(CONTAINER_ID)).thenReturn(container); + resolver = new ClosedContainerReadResolver(mock(ContainerDispatcher.class), containerController, datanode); + } + + @Test + void resolvesReadBlockForClosedContainer() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNotNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @ParameterizedTest + @EnumSource(mode = EnumSource.Mode.EXCLUDE, names = "CLOSED") + void declinesReadBlockForContainerInOtherState(State state) throws IOException { + when(container.getContainerState()).thenReturn(state); + + assertNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @Test + void declinesReadBlockForMissingContainer() throws IOException { + when(containerController.getContainer(CONTAINER_ID)).thenReturn(null); + + assertNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @Test + void declinesOtherCommandForClosedContainer() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNull(resolver.resolve(newRequest(Type.GetBlock, datanodeUuid))); + } + + @Test + void declinesRequestForAnotherDatanode() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNull(resolver.resolve(newRequest(Type.ReadBlock, UUID.randomUUID().toString()))); + } + + private static RaftClientRequest newRequest(Type type, String targetUuid) { + final ContainerCommandRequestProto proto = ContainerCommandRequestProto.newBuilder() + .setCmdType(type) + .setContainerID(CONTAINER_ID) + .setDatanodeUuid(targetUuid) + .build(); + final RaftClientRequest request = mock(RaftClientRequest.class); + when(request.getMessage()).thenReturn(ContainerCommandRequestMessage.toMessage(proto, null)); + when(request.getRaftGroupId()).thenReturn(RaftGroupId.randomId()); + return request; + } +} diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java index 4adb0567e134..9e4bfeecb7d4 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java @@ -19,27 +19,40 @@ import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_BYTES_PER_CHECKSUM_MIN_SIZE; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; import java.io.File; import java.io.InputStream; import java.io.OutputStream; +import java.nio.ByteBuffer; import java.nio.file.Files; import java.security.MessageDigest; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.hdds.StringUtils; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; +import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; import org.apache.hadoop.hdds.scm.OzoneClientConfig; import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.hdds.scm.XceiverClientRatis; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls; import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream; import org.apache.hadoop.hdds.utils.IOUtils; import org.apache.hadoop.hdds.utils.db.CodecBuffer; @@ -51,13 +64,19 @@ import org.apache.hadoop.ozone.client.OzoneClientFactory; import org.apache.hadoop.ozone.client.io.KeyInputStream; import org.apache.hadoop.ozone.client.protocol.ClientProtocol; +import org.apache.hadoop.ozone.container.OzoneTestHelper; import org.apache.hadoop.ozone.container.common.impl.ContainerData; import org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion; import org.apache.hadoop.ozone.om.BucketForTesting; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.ozone.test.GenericTestUtils; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.client.impl.ClientProtoUtils; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto; +import org.apache.ratis.protocol.DataStreamReply; import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.ReferenceCountedObject; import org.apache.ratis.util.SizeInBytes; import org.apache.ratis.util.function.CheckedBiConsumer; import org.junit.jupiter.api.AfterAll; @@ -224,6 +243,87 @@ void runTestReadKey(SizeInBytes keySize, SizeInBytes bytesPerChecksum) throws Ex } } + /** + * Reads a block on a Ratis read-only data stream: first from the open container with the Raft group of its pipeline, + * and then from the closed container with a read pipeline from SCM. No Raft group has the ID of the read pipeline. + */ + @Test + void testRatisStreamReadBlock() throws Exception { + final OzoneConfiguration conf = cluster.getConf(); + final SizeInBytes keySize = SizeInBytes.valueOf("8M"); + try (OzoneClient client = OzoneClientFactory.getRpcClient(conf)) { + final BucketForTesting testBucket = BucketForTesting.newBuilder(client).build(); + final String keyName = "ratisStreamReadKey"; + createKey(testBucket.delegate(), keyName, keySize, SizeInBytes.ONE_MB); + + final OmKeyInfo info = client.getProxy().getKeyInfo( + testBucket.delegate().getVolumeName(), testBucket.delegate().getName(), keyName, false); + final List locations = info.getLatestVersionLocations().createLocationList(); + assertEquals(1, locations.size()); + final BlockID blockId = locations.get(0).getBlockID(); + final Pipeline pipeline = locations.get(0).getPipeline(); + final ContainerData containerData = cluster.getHddsDatanodes().get(0).getDatanodeStateMachine().getContainer() + .getContainerSet().getContainer(blockId.getContainerID()).getContainerData(); + final byte[] expected = Files.readAllBytes( + ContainerLayoutVersion.FILE_PER_BLOCK.getChunkFile(containerData, blockId, null).toPath()); + assertEquals(keySize.getSize(), expected.length); + + assertArrayEquals(expected, ratisStreamRead(pipeline, blockId, expected.length, conf)); + + OzoneTestHelper.waitForContainerClose(cluster, blockId.getContainerID()); + OzoneTestHelper.waitForScmContainerState(cluster, blockId.getContainerID(), LifeCycleState.CLOSED); + final Pipeline readPipeline = cluster.getStorageContainerManager().getClientProtocolServer() + .getContainerWithPipeline(blockId.getContainerID()).getPipeline(); + assertNotEquals(pipeline.getId(), readPipeline.getId()); + + assertArrayEquals(expected, ratisStreamRead(readPipeline, blockId, expected.length, conf)); + } + } + + /** Reads a block on a Ratis read-only data stream to the closest datanode of the pipeline. */ + private static byte[] ratisStreamRead(Pipeline pipeline, BlockID blockId, long length, OzoneConfiguration conf) + throws Exception { + final ContainerCommandRequestProto request = ContainerProtocolCalls.buildReadBlockCommandProto( + blockId, 0, length, CHUNK_SIZE, null, pipeline); + final ByteBuffer message = ContainerCommandRequestMessage.toMessage(request, null).getContent() + .asReadOnlyByteBuffer(); + final ByteArrayOutputStream data = new ByteArrayOutputStream(); + final XceiverClientRatis client = XceiverClientRatis.newXceiverClientRatis(pipeline, conf); + try { + client.connect(); + try (DataStreamInput in = client.getDataStreamApi().streamReadOnly(message)) { + for (;;) { + final ReferenceCountedObject ref = in.readAsync().get(1, TimeUnit.MINUTES); + try { + final DataStreamReply reply = ref.get(); + if (reply.getType() == DataStreamPacketHeaderProto.Type.STREAM_HEADER) { + // The terminal reply + assertTrue(reply.isSuccess(), () -> "Failed: " + ClientProtoUtils.getRaftClientReply(reply)); + return data.toByteArray(); + } + + // [metadata length][the ReadBlock response without its data][data] + final ByteBuffer buffer = reply.nioBuffer(); + final int metadataLength = buffer.getInt(); + final ByteBuffer metadata = buffer.slice(); + metadata.limit(metadataLength); + final ContainerCommandResponseProto response = ContainerCommandResponseProto.parseFrom(metadata); + assertEquals(Result.SUCCESS, response.getResult(), response::getMessage); + assertEquals(data.size(), response.getReadBlock().getOffset()); + buffer.position(buffer.position() + metadataLength); + final byte[] bytes = new byte[buffer.remaining()]; + buffer.get(bytes); + data.write(bytes); + } finally { + ref.release(); + } + } + } + } finally { + client.close(); + } + } + private static void assertClientBytesPerChecksum(SizeInBytes configured, OzoneClient client) { int expected = Math.max(configured.getSizeInt(), OZONE_CLIENT_BYTES_PER_CHECKSUM_MIN_SIZE); OzoneClientConfig clientConfig = client.getConfiguration().getObject(OzoneClientConfig.class); From 394ce6918d90a22fd5ca6dd9fd82c26d410e4406 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Tue, 6 Oct 2026 11:10:46 +0800 Subject: [PATCH 2/7] HDDS-16722. Do not add a DatanodeVersion before the client side exists --- .../src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java index f887a758065e..560a1edf123b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java @@ -38,9 +38,6 @@ public enum DatanodeVersion implements ComponentVersion { SHORT_CIRCUIT_READS(4, "Version with short-circuit read support."), - RATIS_DATASTREAM_READ_BLOCK_SUPPORT(5, "This version supports ReadBlock on Ratis read-only data streams, " - + "also for closed containers without a Raft group."), - FUTURE_VERSION(-1, "Used internally in the client when the server side is " + " newer and an unknown server version has arrived to the client."); From a4d95fad1b486bb7c5a9ebc3e0e28767c3e63a15 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 9 Oct 2026 18:03:52 +0800 Subject: [PATCH 3/7] HDDS-16722. Return empty Ratis parameters instead of null when TLS is off --- .../common/transport/server/ratis/XceiverServerRatis.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java index 2ea583a6e510..dc2991a10667 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java @@ -526,9 +526,6 @@ public static XceiverServerRatis newXceiverServerRatis(HddsDatanodeService hddsD CertificateClient caClient, StateContext context) throws IOException { Parameters parameters = createTlsParameters( new SecurityConfig(ozoneConf), caClient); - if (parameters == null) { - parameters = new Parameters(); - } RaftServerConfigKeys.DataStream.setServerApiResolver(parameters, new ClosedContainerReadResolver(dispatcher, containerController, datanodeDetails)); @@ -554,7 +551,7 @@ private static Parameters createTlsParameters(SecurityConfig conf, return RatisHelper.setServerTlsConf(serverConfig, clientConfig); } - return null; + return new Parameters(); } @Override From 47bdca713e4a3e0103d0fd6ee19aafb0dcbb7908 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 9 Oct 2026 18:12:23 +0800 Subject: [PATCH 4/7] HDDS-16722. Do not scan the container when a ReadBlock stream is cancelled When a reply cannot be written, the Ratis read stream now cancels the read with a CANCELLED status. KeyValueHandler.readBlock passes a CANCELLED status on instead of turning it into CONTAINER_INTERNAL_ERROR, which made the dispatcher scan the container. This also covers the gRPC streaming read, whose onNext throws CANCELLED once the client has gone away. --- .../server/ratis/ContainerStateMachine.java | 15 ++-- .../container/keyvalue/KeyValueHandler.java | 5 ++ .../common/impl/TestHddsDispatcher.java | 84 ++++++++++++++++++- .../ratis/ContainerStateMachineTests.java | 10 +-- 4 files changed, 99 insertions(+), 15 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index 561cbb08858f..732b67b941b3 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -28,7 +28,6 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; -import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; import java.nio.file.Files; @@ -114,6 +113,8 @@ import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException; import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat; import org.apache.ratis.thirdparty.com.google.protobuf.UnsafeByteOperations; +import org.apache.ratis.thirdparty.io.grpc.Status; +import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.apache.ratis.util.FileUtils; import org.apache.ratis.util.JavaUtils; @@ -928,16 +929,12 @@ static long streamReadBlock(ContainerDispatcher dispatcher, ContainerCommandRequ final StreamObserver observer = new StreamObserver() { @Override public void onNext(ContainerCommandResponseProto response) { - if (error.get() != null) { - // The read already failed. Drop the error response that the dispatcher sends after a failed write. - return; - } try { bytesWritten.addAndGet(writeReadBlockReply(stream, response)); } catch (IOException e) { error.set(e); - // Throw to stop reading the block - throw new UncheckedIOException(e); + // Stop reading the block. The handler passes a cancelled stream on, so the container is not scanned. + throw Status.CANCELLED.withDescription("Failed to write a ReadBlock reply").withCause(e).asRuntimeException(); } } @@ -953,8 +950,8 @@ public void onCompleted() { try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) { dispatcher.streamDataReadOnly(request, observer, blockFile, null); - } catch (UncheckedIOException e) { - error.compareAndSet(null, e.getCause()); + } catch (StatusRuntimeException e) { + error.compareAndSet(null, e); } if (error.get() != null) { throw new IOException("Failed to read block " + BlockID.getFromProtobuf(request.getReadBlock().getBlockID()), diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index b4342fcdc915..fb6221ee2adb 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -2312,6 +2312,11 @@ public ContainerCommandResponseProto readBlock( "Failed to readBlock " + request.getReadBlock(), ioe, IO_EXCEPTION); responseProto = ContainerUtils.logAndReturnError(LOG, sce, request); } catch (Exception e) { + if (Status.fromThrowable(e).getCode() == Status.Code.CANCELLED) { + // The stream was cancelled, for example because the client went away. This is not a container problem, so + // pass it on instead of returning an error response, which would make the dispatcher scan the container. + throw e; + } final StorageContainerException sce = new StorageContainerException( "Failed to readBlock " + request.getReadBlock(), e, CONTAINER_INTERNAL_ERROR); LOG.error("", sce); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java index f6137361ca97..ad65989bd0a2 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java @@ -28,9 +28,11 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -43,8 +45,10 @@ import java.io.IOException; import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; @@ -68,6 +72,7 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.hdds.security.token.TokenVerifier; +import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.common.Checksum; import org.apache.hadoop.ozone.common.ChecksumData; @@ -96,12 +101,18 @@ import org.apache.hadoop.ozone.container.keyvalue.ContainerLayoutTestInfo; import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer; import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; +import org.apache.hadoop.ozone.container.ozoneimpl.OnDemandContainerScanner; import org.apache.hadoop.security.token.Token; import org.apache.ozone.test.GenericTestUtils.LogCapturer; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; +import org.apache.ratis.thirdparty.io.grpc.Status; +import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; +import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -639,6 +650,73 @@ public void testDuplicateWriteChunkAndPutBlockRequest() throws IOException { } } + /** + * A ReadBlock whose stream is cancelled, for example because the client went away, is not a container problem, so + * the datanode must not scan the container. Any other failure while reading still makes it scan. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testReadBlockScansContainerOnlyIfTheStreamIsNotCancelled(boolean cancelled) throws Exception { + String testDirPath = testDir.getPath(); + try { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(HDDS_DATANODE_DIR_KEY, testDirPath); + conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, testDirPath); + DatanodeDetails dd = randomDatanodeDetails(); + ContainerSet containerSet = newContainerSet(); + OnDemandContainerScanner scanner = mock(OnDemandContainerScanner.class); + containerSet.registerOnDemandScanner(scanner); + HddsDispatcher hddsDispatcher = createDispatcher(dd, UUID.randomUUID(), conf, null, containerSet); + ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest(dd.getUuidString(), 1L, 1L); + assertEquals(ContainerProtos.Result.SUCCESS, hddsDispatcher.dispatch(writeChunkRequest, null).getResult()); + assertEquals(ContainerProtos.Result.SUCCESS, + hddsDispatcher.dispatch(ContainerTestHelper.getPutBlockRequest(writeChunkRequest), null).getResult()); + ContainerCommandRequestProto readBlockRequest = ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.ReadBlock) + .setContainerID(writeChunkRequest.getContainerID()) + .setDatanodeUuid(dd.getUuidString()) + .setReadBlock(ContainerProtos.ReadBlockRequestProto.newBuilder() + .setBlockID(writeChunkRequest.getWriteChunk().getBlockID()) + .setOffset(0) + .setLength(writeChunkRequest.getWriteChunk().getChunkData().getLen())) + .build(); + + // The stream fails on the data response, and takes the error response that the dispatcher may send after it. + RuntimeException failure = cancelled ? Status.CANCELLED.asRuntimeException() : new IllegalStateException(); + List errorResponses = new ArrayList<>(); + StreamObserver stream = new StreamObserver() { + @Override + public void onNext(ContainerCommandResponseProto response) { + if (response.hasReadBlock()) { + throw failure; + } + errorResponses.add(response); + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) { + if (cancelled) { + assertSame(failure, assertThrows(StatusRuntimeException.class, + () -> hddsDispatcher.streamDataReadOnly(readBlockRequest, stream, blockFile, null))); + } else { + hddsDispatcher.streamDataReadOnly(readBlockRequest, stream, blockFile, null); + } + } + + verify(scanner, times(cancelled ? 0 : 1)).scanContainer(any(), anyString()); + assertEquals(cancelled ? 0 : 1, errorResponses.size()); + } finally { + ContainerMetrics.remove(); + } + } + /** * Creates HddsDispatcher instance with given infos. * @param dd datanode detail info. @@ -654,7 +732,11 @@ static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId, static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId, OzoneConfiguration conf, TokenVerifier tokenVerifier) throws IOException { - ContainerSet containerSet = newContainerSet(); + return createDispatcher(dd, scmId, conf, tokenVerifier, newContainerSet()); + } + + static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId, OzoneConfiguration conf, + TokenVerifier tokenVerifier, ContainerSet containerSet) throws IOException { MutableVolumeSet volumeSet = new MutableVolumeSet(dd.getUuidString(), conf, null, StorageVolume.VolumeType.DATA_VOLUME, null); volumeSet.getVolumesList().stream().forEach(v -> { diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java index 2466b38b0eaf..d48b775b87c5 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java @@ -36,7 +36,6 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.IOException; -import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; import java.util.ArrayList; @@ -319,9 +318,11 @@ public void testTransferToReadBlockReadFailure() throws Exception { @Test public void testTransferToReadBlockWriteFailure() throws Exception { setUpMockReadBlock(observer -> { - // The failed write stops the read. Then the dispatcher sends the error response of the read. - assertThrows(UncheckedIOException.class, () -> observer.onNext(readBlockResponse(0, ByteString.EMPTY))); - observer.onNext(errorResponse()); + final StatusRuntimeException e = assertThrows(StatusRuntimeException.class, + () -> observer.onNext(readBlockResponse(0, ByteString.EMPTY))); + // KeyValueHandler passes a CANCELLED status on, so the dispatcher does not scan the container. + assertEquals(Status.Code.CANCELLED, e.getStatus().getCode()); + throw e; }); final IOException failure = new IOException("Connection closed"); final WritableByteChannel stream = mock(WritableByteChannel.class); @@ -331,7 +332,6 @@ public void testTransferToReadBlockWriteFailure() throws Exception { () -> stateMachine.transferTo(newRequest(ContainerProtos.Type.ReadBlock), stream)); assertSame(failure, e.getCause()); - // The error response is not written to the failed stream. verify(stream, times(1)).write(any()); verify(stream, never()).close(); } From 7fb1932744033fb97d1a4854d7bbd6d4dbe42a33 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 9 Oct 2026 18:12:24 +0800 Subject: [PATCH 5/7] HDDS-16722. Serve reads without a Raft group for every container that is not open or closing SCM gives every container that is not open or closing a read pipeline with a random ID, so the resolver now serves those containers instead of only CLOSED ones. --- .../ratis/ClosedContainerReadResolver.java | 15 +++++------ .../TestClosedContainerReadResolver.java | 27 ++++++++++++------- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java index 6692305d672d..e75c4b195f85 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java @@ -17,8 +17,6 @@ package org.apache.hadoop.ozone.container.common.transport.server.ratis; -import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED; - import java.io.IOException; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; @@ -31,9 +29,9 @@ import org.apache.ratis.server.api.DataStreamApi; /** - * Serves ReadBlock on a Ratis read-only data stream for a closed container, without a Raft group. - * SCM gives the read pipeline of a closed container a random ID, and no Raft group has this ID. - * A closed container does not change, so this datanode can serve the read from its replica. + * Serves ReadBlock on a Ratis read-only data stream, without a Raft group, for a container that is not open or closing. + * SCM gives the read pipeline of such a container a random ID, and no Raft group has this ID. + * The container no longer changes through Raft, so this datanode can serve the read from its replica. */ final class ClosedContainerReadResolver implements DataStreamApi.Resolver { private final ContainerDispatcher dispatcher; @@ -48,8 +46,8 @@ final class ClosedContainerReadResolver implements DataStreamApi.Resolver { } /** - * @return the API to serve a ReadBlock of a closed container on this datanode. Otherwise, null: Ratis then serves - * the request with its Raft group. + * @return the API to serve a ReadBlock of a container on this datanode that is not open or closing. Otherwise, null: + * Ratis then serves the request with its Raft group. */ @Override public DataStreamApi resolve(RaftClientRequest request) throws IOException { @@ -61,7 +59,8 @@ public DataStreamApi resolve(RaftClientRequest request) throws IOException { } final Container container = containerController.getContainer(requestProto.getContainerID()); - if (container == null || container.getContainerState() != CLOSED) { + // Like SCM, which keeps the pipeline with the Raft group for a container that is open or closing + if (container == null || container.getContainerData().isOpen() || container.getContainerData().isClosing()) { return null; } return (message, stream) -> ContainerStateMachine.streamReadBlock(dispatcher, requestProto, stream); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java index 53d4a805197e..f6e84c183b0f 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java @@ -19,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -29,8 +30,10 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; +import org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion; import org.apache.hadoop.ozone.container.common.interfaces.Container; import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; import org.apache.ratis.protocol.RaftClientRequest; import org.apache.ratis.protocol.RaftGroupId; @@ -46,7 +49,7 @@ class TestClosedContainerReadResolver { private static final long CONTAINER_ID = 1L; private ContainerController containerController; - private Container container; + private KeyValueContainerData containerData; private ClosedContainerReadResolver resolver; private String datanodeUuid; @@ -56,22 +59,26 @@ void setUp() { final DatanodeDetails datanode = mock(DatanodeDetails.class); when(datanode.getUuidString()).thenReturn(datanodeUuid); containerController = mock(ContainerController.class); - container = mock(Container.class); + containerData = new KeyValueContainerData(CONTAINER_ID, ContainerLayoutVersion.FILE_PER_BLOCK, 1L << 30, + UUID.randomUUID().toString(), datanodeUuid); + final Container container = mock(Container.class); + doReturn(containerData).when(container).getContainerData(); when(containerController.getContainer(CONTAINER_ID)).thenReturn(container); resolver = new ClosedContainerReadResolver(mock(ContainerDispatcher.class), containerController, datanode); } - @Test - void resolvesReadBlockForClosedContainer() throws IOException { - when(container.getContainerState()).thenReturn(State.CLOSED); + @ParameterizedTest + @EnumSource(mode = EnumSource.Mode.EXCLUDE, names = {"OPEN", "CLOSING"}) + void resolvesReadBlockForContainerThatIsNotOpen(State state) throws IOException { + containerData.setState(state); assertNotNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); } @ParameterizedTest - @EnumSource(mode = EnumSource.Mode.EXCLUDE, names = "CLOSED") - void declinesReadBlockForContainerInOtherState(State state) throws IOException { - when(container.getContainerState()).thenReturn(state); + @EnumSource(names = {"OPEN", "CLOSING"}) + void declinesReadBlockForOpenContainer(State state) throws IOException { + containerData.setState(state); assertNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); } @@ -85,14 +92,14 @@ void declinesReadBlockForMissingContainer() throws IOException { @Test void declinesOtherCommandForClosedContainer() throws IOException { - when(container.getContainerState()).thenReturn(State.CLOSED); + containerData.setState(State.CLOSED); assertNull(resolver.resolve(newRequest(Type.GetBlock, datanodeUuid))); } @Test void declinesRequestForAnotherDatanode() throws IOException { - when(container.getContainerState()).thenReturn(State.CLOSED); + containerData.setState(State.CLOSED); assertNull(resolver.resolve(newRequest(Type.ReadBlock, UUID.randomUUID().toString()))); } From f6a65910c328fcf33922a04536e05f5e74aa03e1 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 9 Oct 2026 18:55:09 +0800 Subject: [PATCH 6/7] HDDS-16722. Pass stream status exceptions on in a catch branch of their own Also say in the resolver why it checks isClosing, and in the Ratis read stream why it throws CANCELLED. --- .../server/ratis/ClosedContainerReadResolver.java | 3 ++- .../transport/server/ratis/ContainerStateMachine.java | 3 ++- .../ozone/container/keyvalue/KeyValueHandler.java | 11 ++++++----- 3 files changed, 10 insertions(+), 7 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java index e75c4b195f85..e16a82fe8dd8 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java @@ -59,7 +59,8 @@ public DataStreamApi resolve(RaftClientRequest request) throws IOException { } final Container container = containerController.getContainer(requestProto.getContainerID()); - // Like SCM, which keeps the pipeline with the Raft group for a container that is open or closing + // The states for which SCM's ContainerInfo#isOpen is true, so SCM keeps the pipeline with the Raft group. + // On the datanode, ContainerData#isOpen covers only OPEN. if (container == null || container.getContainerData().isOpen() || container.getContainerData().isClosing()) { return null; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index 22b9711ee281..a112ae74d5aa 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -936,7 +936,8 @@ public void onNext(ContainerCommandResponseProto response) { bytesWritten.addAndGet(writeReadBlockReply(stream, response)); } catch (IOException e) { error.set(e); - // Stop reading the block. The handler passes a cancelled stream on, so the container is not scanned. + // Stop reading the block, as the onNext of a cancelled gRPC call does by throwing CANCELLED. + // The handler passes the status on, so the container is not scanned. throw Status.CANCELLED.withDescription("Failed to write a ReadBlock reply").withCause(e).asRuntimeException(); } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 1a7b1e65ff5c..7bd801beda6c 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -175,6 +175,7 @@ import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.thirdparty.io.grpc.Status; +import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException; import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.apache.ratis.util.function.CheckedConsumer; import org.slf4j.Logger; @@ -2316,12 +2317,12 @@ public ContainerCommandResponseProto readBlock( final StorageContainerException sce = new StorageContainerException( "Failed to readBlock " + request.getReadBlock(), ioe, IO_EXCEPTION); responseProto = ContainerUtils.logAndReturnError(LOG, sce, request); + } catch (StatusRuntimeException e) { + // The response stream failed, for example with CANCELLED because the client went away. This is not a container + // problem, so pass it on instead of returning an error response, which would make the dispatcher scan the + // container. + throw e; } catch (Exception e) { - if (Status.fromThrowable(e).getCode() == Status.Code.CANCELLED) { - // The stream was cancelled, for example because the client went away. This is not a container problem, so - // pass it on instead of returning an error response, which would make the dispatcher scan the container. - throw e; - } final StorageContainerException sce = new StorageContainerException( "Failed to readBlock " + request.getReadBlock(), e, CONTAINER_INTERNAL_ERROR); LOG.error("", sce); From a2734952143f930b186657b4cd228738c91a5e00 Mon Sep 17 00:00:00 2001 From: peterxcli Date: Fri, 9 Oct 2026 19:07:40 +0800 Subject: [PATCH 7/7] HDDS-16722. Remove three code comments --- .../transport/server/ratis/ClosedContainerReadResolver.java | 2 -- .../common/transport/server/ratis/ContainerStateMachine.java | 2 -- .../hadoop/ozone/container/keyvalue/KeyValueHandler.java | 3 --- 3 files changed, 7 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java index e16a82fe8dd8..4a5e20c3fda4 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java @@ -59,8 +59,6 @@ public DataStreamApi resolve(RaftClientRequest request) throws IOException { } final Container container = containerController.getContainer(requestProto.getContainerID()); - // The states for which SCM's ContainerInfo#isOpen is true, so SCM keeps the pipeline with the Raft group. - // On the datanode, ContainerData#isOpen covers only OPEN. if (container == null || container.getContainerData().isOpen() || container.getContainerData().isClosing()) { return null; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index a112ae74d5aa..890d3920e74f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -936,8 +936,6 @@ public void onNext(ContainerCommandResponseProto response) { bytesWritten.addAndGet(writeReadBlockReply(stream, response)); } catch (IOException e) { error.set(e); - // Stop reading the block, as the onNext of a cancelled gRPC call does by throwing CANCELLED. - // The handler passes the status on, so the container is not scanned. throw Status.CANCELLED.withDescription("Failed to write a ReadBlock reply").withCause(e).asRuntimeException(); } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 7bd801beda6c..083aa3cde57a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -2318,9 +2318,6 @@ public ContainerCommandResponseProto readBlock( "Failed to readBlock " + request.getReadBlock(), ioe, IO_EXCEPTION); responseProto = ContainerUtils.logAndReturnError(LOG, sce, request); } catch (StatusRuntimeException e) { - // The response stream failed, for example with CANCELLED because the client went away. This is not a container - // problem, so pass it on instead of returning an error response, which would make the dispatcher scan the - // container. throw e; } catch (Exception e) { final StorageContainerException sce = new StorageContainerException(