This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 247f5cab1f6 HDDS-15758. Commit PutBlock without Raft in Client (#10823)
247f5cab1f6 is described below
commit 247f5cab1f6f7b15284bb51c09d699836e385551
Author: Rui Wang <[email protected]>
AuthorDate: Thu Jul 30 00:31:17 2026 +0800
HDDS-15758. Commit PutBlock without Raft in Client (#10823)
---
.../apache/hadoop/hdds/scm/OzoneClientConfig.java | 16 +++++++
.../hdds/scm/storage/BlockDataStreamOutput.java | 28 ++++++++++-
.../hadoop/hdds/scm/TestOzoneClientConfig.java | 17 +++++++
.../hdds/scm/storage/MockDatanodePipeline.java | 29 +++++++++++-
.../scm/storage/TestBlockDataStreamOutput.java | 22 ++++++++-
.../java/org/apache/hadoop/hdds/HddsUtils.java | 1 +
.../org/apache/hadoop/ozone/audit/DNAction.java | 1 +
.../container/common/impl/HddsDispatcher.java | 4 +-
.../ozone/container/keyvalue/KeyValueHandler.java | 1 +
.../container/keyvalue/impl/BlockManagerImpl.java | 14 ++++++
.../client/rpc/TestBlockDataStreamOutput.java | 55 +++++++++++++++-------
.../rpc/TestContainerStateMachineStream.java | 30 +++++++++---
12 files changed, 187 insertions(+), 31 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
index dba10525d2b..70eae2073ff 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java
@@ -289,6 +289,14 @@ public class OzoneClientConfig {
tags = ConfigTag.CLIENT)
private boolean enablePutblockPiggybacking = false;
+ @Config(key = "ozone.client.datastream.putblock.on.close.enabled",
+ defaultValue = "false",
+ type = ConfigType.BOOLEAN,
+ description = "When enabled, use StreamInitWithPutBlock so datanodes
commit PutBlock " +
+ "when the Ratis data stream closes instead of via a separate
WriteAsync PutBlock.",
+ tags = ConfigTag.CLIENT)
+ private boolean datastreamPutBlockOnCloseEnabled = false;
+
@Config(key = "ozone.client.key.write.concurrency",
defaultValue = "1",
description = "Maximum concurrent writes allowed on each key. " +
@@ -646,6 +654,14 @@ public void setStreamReadTimeout(Duration
streamReadTimeout) {
this.streamReadTimeout = streamReadTimeout;
}
+ public boolean isDatastreamPutBlockOnCloseEnabled() {
+ return datastreamPutBlockOnCloseEnabled;
+ }
+
+ public void setDatastreamPutBlockOnCloseEnabled(boolean
datastreamPutBlockOnCloseEnabled) {
+ this.datastreamPutBlockOnCloseEnabled = datastreamPutBlockOnCloseEnabled;
+ }
+
/**
* Enum for indicating what mode to use when combining chunk and block
* checksums to define an aggregate FileChecksum. This should be considered
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
index 1ababc7a1d7..e3fff7528d4 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
@@ -211,9 +211,12 @@ private DataStreamOutput setupStream(Pipeline pipeline)
throws IOException {
// TODO: The datanode UUID is not used meaningfully, consider deprecating
// it or remove it completely if possible
String id = pipeline.getFirstNode().getUuidString();
+ ContainerProtos.Type streamInitType =
config.isDatastreamPutBlockOnCloseEnabled()
+ ? ContainerProtos.Type.StreamInitWithPutBlock
+ : ContainerProtos.Type.StreamInit;
ContainerProtos.ContainerCommandRequestProto.Builder builder =
ContainerProtos.ContainerCommandRequestProto.newBuilder()
- .setCmdType(ContainerProtos.Type.StreamInit)
+ .setCmdType(streamInitType)
.setContainerID(blockID.get().getContainerID())
.setDatanodeUuid(id).setWriteChunk(writeChunkRequest);
@@ -416,6 +419,10 @@ public void executePutBlock(boolean close,
byteBufferList = null;
}
waitFuturesComplete();
+ if (close && config.isDatastreamPutBlockOnCloseEnabled()) {
+ // Wait for boundary PutBlock(s) before appending the stream-close
PutBlock.
+ waitPutBlockFuturesComplete();
+ }
final BlockData blockData = containerBlockData.build();
if (close) {
// HDDS-12007 changed datanodes to ignore the following PutBlock request.
@@ -437,8 +444,12 @@ public void executePutBlock(boolean close,
}
}
});
+ if (config.isDatastreamPutBlockOnCloseEnabled()) {
+ // PutBlock is supposed to be committed after the data stream close so
there
+ // is no need to continue.
+ return;
+ }
}
-
try {
XceiverClientReply asyncReply =
putBlockAsync(xceiverClient, blockData, close, tokenString);
@@ -545,6 +556,19 @@ public void waitFuturesComplete() throws IOException {
}
}
+ private void waitPutBlockFuturesComplete() throws IOException {
+ if (putBlockFutures.isEmpty()) {
+ return;
+ }
+ try {
+
CompletableFuture.allOf(putBlockFutures.toArray(EMPTY_FUTURE_ARRAY)).get();
+ checkOpen();
+ } catch (Exception e) {
+ LOG.warn("Failed to commit PutBlock before stream close: " + e);
+ throw new IOException(e);
+ }
+ }
+
/**
* @param close whether the flush is happening as part of closing the stream
*/
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestOzoneClientConfig.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestOzoneClientConfig.java
index 5c1eeccff91..cb05d31a2d6 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestOzoneClientConfig.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestOzoneClientConfig.java
@@ -91,4 +91,21 @@ public void testStreamReadConfigParsing() {
assertEquals(2 << 20, clientConfig.getStreamReadResponseDataSize());
assertEquals(Duration.ofSeconds(5), clientConfig.getStreamReadTimeout());
}
+
+ @Test
+ void testDatastreamPutBlockOnCloseEnabledDefault() {
+ OzoneClientConfig subject = new OzoneConfiguration()
+ .getObject(OzoneClientConfig.class);
+ assertFalse(subject.isDatastreamPutBlockOnCloseEnabled());
+ }
+
+ @Test
+ void testDatastreamPutBlockOnCloseConfigParsing() {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setBoolean("ozone.client.datastream.putblock.on.close.enabled", true);
+
+ OzoneClientConfig subject = conf.getObject(OzoneClientConfig.class);
+
+ assertTrue(subject.isDatastreamPutBlockOnCloseEnabled());
+ }
}
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/MockDatanodePipeline.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/MockDatanodePipeline.java
index 85341fd2721..c1c11b0d1a8 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/MockDatanodePipeline.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/MockDatanodePipeline.java
@@ -43,6 +43,7 @@
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.PutBlockResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
+import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage;
import org.apache.hadoop.hdds.scm.XceiverClientFactory;
import org.apache.hadoop.hdds.scm.XceiverClientManager;
import org.apache.hadoop.hdds.scm.XceiverClientRatis;
@@ -56,6 +57,7 @@
import org.apache.ratis.io.WriteOption;
import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.RoutingTable;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
/**
* A stateful test harness that simulates a datanode pipeline for {@link
BlockDataStreamOutput} unit tests.
@@ -76,6 +78,7 @@ public class MockDatanodePipeline {
private final List<byte[]> receivedChunks = Collections.synchronizedList(new
ArrayList<>());
private final List<ContainerCommandRequestProto> receivedPutBlocks =
Collections.synchronizedList(new ArrayList<>());
private final AtomicInteger watchForCommitCount = new AtomicInteger(0);
+ private volatile Type streamInitType;
// Commit tracking
private final AtomicLong nextLogIndex = new AtomicLong(1);
@@ -120,8 +123,14 @@ public MockDatanodePipeline(BlockID blockID) throws
IOException {
// Both overloads must be stubbed: stream(ByteBuffer) and
stream(ByteBuffer, RoutingTable) — the pipeline-mode
// default is true, so the 2-arg overload is what
BlockDataStreamOutput.setupStream calls.
DataStreamApi dataStreamApi = mock(DataStreamApi.class);
-
doReturn(mockDataStreamOutput).when(dataStreamApi).stream(any(ByteBuffer.class));
-
doReturn(mockDataStreamOutput).when(dataStreamApi).stream(any(ByteBuffer.class),
any(RoutingTable.class));
+ doAnswer(invocation -> {
+ captureStreamInitType(invocation.getArgument(0));
+ return mockDataStreamOutput;
+ }).when(dataStreamApi).stream(any(ByteBuffer.class));
+ doAnswer(invocation -> {
+ captureStreamInitType(invocation.getArgument(0));
+ return mockDataStreamOutput;
+ }).when(dataStreamApi).stream(any(ByteBuffer.class),
any(RoutingTable.class));
doReturn(dataStreamApi).when(xceiverClient).getDataStreamApi();
// Setup sendCommandAsync (putBlock) behavior
@@ -229,6 +238,10 @@ public int getWatchForCommitCount() {
return watchForCommitCount.get();
}
+ public Type getStreamInitType() {
+ return streamInitType;
+ }
+
/** Concatenate all received chunks into a single byte array. */
public byte[] getAllReceivedData() {
int total = receivedChunks.stream().mapToInt(c -> c.length).sum();
@@ -263,6 +276,18 @@ public MockDatanodePipeline failWatchAfter(int n,
Supplier<Throwable> err) {
// --- Helpers ---
+ private void captureStreamInitType(ByteBuffer buffer) {
+ ByteBuffer dup = buffer.duplicate();
+ byte[] bytes = new byte[dup.remaining()];
+ dup.get(bytes);
+ try {
+ streamInitType = ContainerCommandRequestMessage.toProto(
+ ByteString.copyFrom(bytes), null).getCmdType();
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to decode stream init request",
e);
+ }
+ }
+
private static ContainerCommandResponseProto buildPutBlockResponse(BlockID
blockID) {
return ContainerCommandResponseProto.newBuilder()
.setCmdType(Type.PutBlock)
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java
index 7634328c56f..8f51e3c9a41 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java
@@ -30,8 +30,11 @@
import java.util.concurrent.CompletionException;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/**
* Unit tests for {@link BlockDataStreamOutput} exercised through the {@link
ByteBufferStreamOutput} interface with a
@@ -59,16 +62,33 @@ private static OzoneClientConfig createConfig() {
}
private BlockDataStreamOutput createStream(MockDatanodePipeline pipeline)
throws IOException {
+ return createStream(pipeline, createConfig());
+ }
+
+ private BlockDataStreamOutput createStream(
+ MockDatanodePipeline pipeline, OzoneClientConfig config) throws
IOException {
List<StreamBuffer> bufferList = new ArrayList<>();
return new BlockDataStreamOutput(
pipeline.getBlockID(),
pipeline.getClientFactory(),
pipeline.getPipeline(),
- createConfig(),
+ config,
null, // no token
bufferList);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void streamInitTypeFollowsClientConfig(boolean putBlockOnCloseEnabled)
throws Exception {
+ MockDatanodePipeline pipeline = new MockDatanodePipeline();
+ OzoneClientConfig config = createConfig();
+ config.setDatastreamPutBlockOnCloseEnabled(putBlockOnCloseEnabled);
+ try (BlockDataStreamOutput stream = createStream(pipeline, config)) {
+ Type expected = putBlockOnCloseEnabled ? Type.StreamInitWithPutBlock :
Type.StreamInit;
+ assertEquals(expected, pipeline.getStreamInitType());
+ }
+ }
+
@Test
void writeSubChunkThenClose() throws Exception {
MockDatanodePipeline pipeline = new MockDatanodePipeline();
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java
index 46289059dbb..3e6715b11c3 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java
@@ -373,6 +373,7 @@ public static boolean isReadOnly(
case PutBlock:
case PutSmallFile:
case StreamInit:
+ case StreamInitWithPutBlock:
case StreamWrite:
case FinalizeBlock:
return false;
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/audit/DNAction.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/audit/DNAction.java
index 61d1c49da04..c5a62d8e79f 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/audit/DNAction.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/audit/DNAction.java
@@ -41,6 +41,7 @@ public enum DNAction implements AuditAction {
CLOSE_CONTAINER,
GET_COMMITTED_BLOCK_LENGTH,
STREAM_INIT,
+ STREAM_INIT_WITH_PUT_BLOCK,
FINALIZE_BLOCK,
ECHO,
GET_CONTAINER_CHECKSUM_INFO,
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
index 47d29832587..1564e3ddbf5 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java
@@ -226,7 +226,8 @@ private ContainerCommandResponseProto dispatchRequest(
(cmdType == Type.WriteChunk && dispatcherContext != null
&& dispatcherContext.getStage()
== DispatcherContext.WriteChunkStage.WRITE_DATA)
- || (cmdType == Type.StreamInit);
+ || (cmdType == Type.StreamInit)
+ || (cmdType == Type.StreamInitWithPutBlock);
boolean isWriteCommitStage =
(cmdType == Type.WriteChunk && dispatcherContext != null
&& dispatcherContext.getStage()
@@ -925,6 +926,7 @@ private static DNAction getAuditAction(Type cmdType) {
case CloseContainer : return DNAction.CLOSE_CONTAINER;
case GetCommittedBlockLength : return DNAction.GET_COMMITTED_BLOCK_LENGTH;
case StreamInit : return DNAction.STREAM_INIT;
+ case StreamInitWithPutBlock: return DNAction.STREAM_INIT_WITH_PUT_BLOCK;
case FinalizeBlock : return DNAction.FINALIZE_BLOCK;
case Echo : return DNAction.ECHO;
case GetContainerChecksumInfo: return DNAction.GET_CONTAINER_CHECKSUM_INFO;
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 18486bc9ef3..66163be780d 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
@@ -355,6 +355,7 @@ static ContainerCommandResponseProto
dispatchRequest(KeyValueHandler handler,
case WriteChunk:
return handler.handleWriteChunk(request, kvContainer, dispatcherContext);
case StreamInit:
+ case StreamInitWithPutBlock:
return handler.handleStreamInit(request, kvContainer, dispatcherContext);
case ListChunk:
return handler.handleUnsupportedOp(request);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
index 62dbcbe808e..46929633d09 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
@@ -209,6 +209,20 @@ public long persistPutBlock(KeyValueContainer container,
// container to determine whether the blockCount is already incremented
// for this block in the DB or not.
long localID = data.getLocalID();
+ // For the PutBlock that is endOfBlock and meanwhile bscId = 0, it means
+ // this PutBlock comes from data stream close without going through the
+ // Raft, thus there is no log index. In this case, we should not let
+ // 0 to overwrite previous possible PutBlocks from Ratis log that were
+ // generated during immediate flushes from the active data stream.
Instead,
+ // we should load the latest bscid and reuse that id.
+ if (endOfBlock && bcsId == 0) {
+ BlockData existing = db.getStore().getBlockDataTable()
+ .get(containerData.getBlockKey(localID));
+ if (existing != null) {
+ bcsId = existing.getBlockCommitSequenceId();
+ data.setBlockCommitSequenceId(bcsId);
+ }
+ }
boolean isBlockInCache =
container.isBlockInPendingPutBlockCache(localID);
boolean incrBlockCount = false;
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
index 3cea0590d85..b4337a494e1 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
@@ -152,8 +152,10 @@ static MiniOzoneCluster createCluster() throws IOException,
private static Stream<Arguments> clientParameters() {
return Stream.of(
- Arguments.of(true),
- Arguments.of(false)
+ Arguments.of(true, true),
+ Arguments.of(true, false),
+ Arguments.of(false, true),
+ Arguments.of(false, false)
);
}
@@ -166,11 +168,19 @@ private static Stream<Arguments> dataLengthParameters() {
);
}
+ private static Stream<Arguments> streamWriteParameters() {
+ return dataLengthParameters().flatMap(dataLength ->
+ Stream.of(true, false).map(putBlockOnCloseEnabled ->
+ Arguments.of(dataLength.get()[0], putBlockOnCloseEnabled)));
+ }
+
static OzoneClientConfig newClientConfig(ConfigurationSource source,
- boolean flushDelay) {
+ boolean flushDelay,
+ boolean putBlockOnCloseEnabled) {
OzoneClientConfig clientConfig = source.getObject(OzoneClientConfig.class);
clientConfig.setChecksumType(ContainerProtos.ChecksumType.NONE);
clientConfig.setStreamBufferFlushDelay(flushDelay);
+ clientConfig.setDatastreamPutBlockOnCloseEnabled(putBlockOnCloseEnabled);
return clientConfig;
}
@@ -198,13 +208,17 @@ public void shutdown() {
}
@ParameterizedTest
- @MethodSource("dataLengthParameters")
+ @MethodSource("streamWriteParameters")
@Flaky("HDDS-12027")
- public void testStreamWrite(int dataLength) throws Exception {
- OzoneClientConfig config = newClientConfig(cluster.getConf(), false);
+ public void testStreamWrite(int dataLength, boolean putBlockOnCloseEnabled)
throws Exception {
+ OzoneClientConfig config = newClientConfig(cluster.getConf(), false,
putBlockOnCloseEnabled);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
testWrite(client, dataLength);
- testWriteWithFailure(client, dataLength);
+ // Forced container close before stream close relies on async PutBlock
recovery;
+ // that path is not used when PutBlock is committed only on data stream
close.
+ if (!putBlockOnCloseEnabled) {
+ testWriteWithFailure(client, dataLength);
+ }
}
}
@@ -253,8 +267,9 @@ static void validateData(OzoneClient client, String
keyName, byte[] data) throws
@ParameterizedTest
@MethodSource("clientParameters")
- public void testPutBlockAtBoundary(boolean flushDelay) throws Exception {
- OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay);
+ public void testPutBlockAtBoundary(boolean flushDelay, boolean
putBlockOnCloseEnabled)
+ throws Exception {
+ OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay,
putBlockOnCloseEnabled);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
int dataLength = 500;
XceiverClientMetrics metrics =
@@ -273,19 +288,21 @@ public void testPutBlockAtBoundary(boolean flushDelay)
throws Exception {
assertThat(metrics.getPendingContainerOpCountMetrics(ContainerProtos.Type.PutBlock))
.isLessThanOrEqualTo(pendingPutBlockCount + 1);
key.close();
- // Since data length is 500 , first putBlock will be at 400(flush
boundary)
- // and the other at 500
+ // Since data length is 500, first putBlock will be at 400 (flush
boundary).
+ // Close commits via WriteAsync PutBlock only when putBlockOnClose is
disabled.
+ int expectedPutBlocks = putBlockOnCloseEnabled ? 1 : 2;
assertEquals(
metrics.getContainerOpCountMetrics(ContainerProtos.Type.PutBlock),
- putBlockCount + 2);
+ putBlockCount + expectedPutBlocks);
validateData(client, keyName, data);
}
}
@ParameterizedTest
@MethodSource("clientParameters")
- public void testMinPacketSize(boolean flushDelay) throws Exception {
- OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay);
+ public void testMinPacketSize(boolean flushDelay, boolean
putBlockOnCloseEnabled)
+ throws Exception {
+ OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay,
putBlockOnCloseEnabled);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
String keyName = getKeyName();
XceiverClientMetrics metrics =
@@ -312,8 +329,9 @@ public void testMinPacketSize(boolean flushDelay) throws
Exception {
@ParameterizedTest
@MethodSource("clientParameters")
- public void testTotalAckDataLength(boolean flushDelay) throws Exception {
- OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay);
+ public void testTotalAckDataLength(boolean flushDelay, boolean
putBlockOnCloseEnabled)
+ throws Exception {
+ OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay,
putBlockOnCloseEnabled);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
int dataLength = 400;
String keyName = getKeyName();
@@ -334,8 +352,9 @@ public void testTotalAckDataLength(boolean flushDelay)
throws Exception {
@ParameterizedTest
@MethodSource("clientParameters")
- public void testDatanodeVersion(boolean flushDelay) throws Exception {
- OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay);
+ public void testDatanodeVersion(boolean flushDelay, boolean
putBlockOnCloseEnabled)
+ throws Exception {
+ OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay,
putBlockOnCloseEnabled);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
// Verify all DNs internally have versions set correctly
List<HddsDatanodeService> dns = cluster.getHddsDatanodes();
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
index 58af0db2b11..8f8bc5cf946 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
@@ -26,11 +26,15 @@
import java.nio.ByteBuffer;
import java.util.List;
import java.util.UUID;
+import java.util.stream.Stream;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.client.ReplicationType;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
import org.apache.hadoop.hdds.utils.IOUtils;
import org.apache.hadoop.ozone.client.ObjectStore;
import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.io.KeyDataStreamOutput;
import org.apache.hadoop.ozone.client.io.OzoneDataStreamOutput;
import org.apache.hadoop.ozone.container.ContainerTestHelper;
@@ -40,7 +44,8 @@
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
/**
* Tests the containerStateMachine stream handling.
@@ -48,7 +53,6 @@
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public abstract class TestContainerStateMachineStream implements
NonHATests.TestCase {
private OzoneClient client;
- private ObjectStore objectStore;
private String volumeName;
private String bucketName;
private int chunkSize;
@@ -58,7 +62,7 @@ void setup() throws Exception {
chunkSize = (int)
cluster().getConf().getStorageSize(OZONE_SCM_CHUNK_SIZE_KEY, 1024 * 1024,
StorageUnit.BYTES);
client = cluster().newClient();
- objectStore = client.getObjectStore();
+ ObjectStore objectStore = client.getObjectStore();
volumeName = "vol-" + UUID.randomUUID();
bucketName = "teststreambucket";
@@ -71,14 +75,26 @@ void shutdown() {
IOUtils.closeQuietly(client);
}
+ private static Stream<Arguments> streamingParameters() {
+ return Stream.of(-1, +1).flatMap(offset ->
+ Stream.of(false, true).map(putBlockOnCloseEnabled ->
+ Arguments.of(offset, putBlockOnCloseEnabled)));
+ }
+
@ParameterizedTest
- @ValueSource(ints = {-1, +1})
- void testContainerStateMachineForStreaming(int offset) throws Exception {
+ @MethodSource("streamingParameters")
+ void testContainerStateMachineForStreaming(int offset, boolean
putBlockOnCloseEnabled)
+ throws Exception {
final int size = chunkSize + offset;
+ OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf());
+ OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+ clientConfig.setDatastreamPutBlockOnCloseEnabled(putBlockOnCloseEnabled);
+ conf.setFromObject(clientConfig);
final List<OmKeyLocationInfo> locationInfoList;
- try (OzoneDataStreamOutput key = createStreamKey("key" + offset,
ReplicationType.RATIS, size,
- objectStore, volumeName, bucketName)) {
+ try (OzoneClient streamingClient = OzoneClientFactory.getRpcClient(conf);
+ OzoneDataStreamOutput key = createStreamKey("key" + offset + "-" +
putBlockOnCloseEnabled,
+ ReplicationType.RATIS, size, streamingClient.getObjectStore(),
volumeName, bucketName)) {
byte[] data = ContainerTestHelper.generateData(size, true);
key.write(ByteBuffer.wrap(data));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
