peterxcli commented on code in PR #10230: URL: https://github.com/apache/ozone/pull/10230#discussion_r3529709475
########## hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.hdds.scm.storage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import org.apache.commons.lang3.RandomUtils; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link BlockDataStreamOutput} exercised through the {@link ByteBufferStreamOutput} interface with a + * mocked datanode pipeline. + */ +class TestBlockDataStreamOutput { + + // Config: CHUNK=100, flush boundary=4 chunks (400B), window=5 chunks (500B) + private static final int CHUNK_SIZE = 100; + private static final long DS_FLUSH_SIZE = 400; + private static final long STREAM_WINDOW = 500; + + private static OzoneClientConfig createConfig() { + OzoneClientConfig config = new OzoneClientConfig(); + config.setDataStreamMinPacketSize(CHUNK_SIZE); + config.setDataStreamBufferFlushSize(DS_FLUSH_SIZE); + config.setStreamWindowSize(STREAM_WINDOW); + config.setStreamBufferSize(CHUNK_SIZE); + config.setStreamBufferFlushSize(DS_FLUSH_SIZE); + config.setStreamBufferMaxSize(2 * DS_FLUSH_SIZE); + config.setStreamBufferFlushDelay(false); + config.setChecksumType(ContainerProtos.ChecksumType.NONE); + config.setBytesPerChecksum(CHUNK_SIZE); + return config; + } + + private BlockDataStreamOutput createStream(FakeDatanodePipeline fake) throws IOException { + List<StreamBuffer> bufferList = new ArrayList<>(); + return new BlockDataStreamOutput( + fake.getBlockID(), + fake.getClientFactory(), + fake.getPipeline(), + createConfig(), + null, // no token + bufferList); + } + + @Test + void writeSubChunkThenClose() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + try (BlockDataStreamOutput stream = createStream(fake)) { + byte[] data = randomBytes(50); + stream.write(ByteBuffer.wrap(data), 0, data.length); + // No chunk shipped yet — data is in currentBuffer + assertEquals(0, fake.getReceivedChunks().size(), "No chunk should be shipped before close for sub-chunk write"); + } + // After close: 1 chunk flushed + 1 putBlock + assertEquals(1, fake.getReceivedChunks().size()); + assertThat(fake.getReceivedPutBlocks()).isNotEmpty(); + } + + @Test + void writeExactChunkThenClose() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + byte[] data = randomBytes(CHUNK_SIZE); + try (BlockDataStreamOutput stream = createStream(fake)) { + stream.write(ByteBuffer.wrap(data), 0, data.length); + // Exact chunk → shipped immediately (currentBuffer full) + assertEquals(1, fake.getReceivedChunks().size()); + } + assertThat(fake.getReceivedPutBlocks()).isNotEmpty(); + assertArrayEquals(data, fake.getAllReceivedData()); + } + + @Test + void writeFlushBoundaryTriggersPutBlock() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + byte[] data = randomBytes(400); + try (BlockDataStreamOutput stream = createStream(fake)) { + stream.write(ByteBuffer.wrap(data), 0, data.length); + // 4 chunks of 100B each → hits flush boundary → 1 putBlock + assertEquals(4, fake.getReceivedChunks().size()); + assertEquals(1, fake.getReceivedPutBlocks().size(), "PutBlock should trigger at flush boundary (400B)"); + } + // Close adds another putBlock + assertEquals(2, fake.getReceivedPutBlocks().size()); + } + + @Test + void writeAcrossStreamWindowTriggersBackPressure() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + // Window = 500B = 5 chunks. Writing 500B should trigger back-pressure. + byte[] data = randomBytes(500); + try (BlockDataStreamOutput stream = createStream(fake)) { + stream.write(ByteBuffer.wrap(data), 0, data.length); + // 5 chunks written, putBlock at 400B boundary, back-pressure at 500B + // should have triggered watchForCommit + assertThat(fake.getWatchForCommitCount()) + .as("watchForCommit should be called for back-pressure") + .isGreaterThanOrEqualTo(1); + } + assertArrayEquals(data, fake.getAllReceivedData()); + } + + @Test + void hsyncFlushesAndWaitsForCommit() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + try (BlockDataStreamOutput stream = createStream(fake)) { + byte[] data = randomBytes(200); + stream.write(ByteBuffer.wrap(data), 0, data.length); + stream.hsync(); + // 2 chunks, 1 putBlock (from hsync), watch called + assertEquals(2, fake.getReceivedChunks().size()); + assertThat(fake.getReceivedPutBlocks()).isNotEmpty(); + assertThat(fake.getWatchForCommitCount()).isGreaterThanOrEqualTo(1); + } + } + + //@Test - skipped as it fails now. + void hsyncPropagatesIOException() throws Exception { + FakeDatanodePipeline fake = new FakeDatanodePipeline(); + // Fail the first putBlock + fake.failPutBlockAfter(0, () -> new IOException("simulated putBlock fail")); + + BlockDataStreamOutput stream = createStream(fake); + byte[] data = randomBytes(200); + stream.write(ByteBuffer.wrap(data), 0, data.length); + + // hsync should propagate the IOException from the failed putBlock + assertThrows(IOException.class, stream::hsync, "hsync() must propagate IOException from failed putBlock"); Review Comment: > if the exception is not exposed, then a client might falsely think that the sync was successful, while it failed, so the client has a different view on what was persisted and what was not. make sense, is there any ticket for this? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
