amaliujia commented on code in PR #11384:
URL: https://github.com/apache/ozone/pull/11384#discussion_r4477501855
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java:
##########
@@ -435,17 +427,18 @@ public void watchForCommit(boolean bufferFull) throws
IOException {
public void executePutBlock(boolean sync, boolean close) throws IOException {
checkOpen();
long flushPos = totalDataFlushedLength;
+ final PendingFlush flush = pendingFlush;
+ pendingFlush = new PendingFlush();
final List<StreamBuffer> byteBufferList;
- if (buffersForPutBlock != null) {
+ if (flush.hasBuffersForPutBlock()) {
Objects.requireNonNull(bufferList, "bufferList == null");
- byteBufferList = buffersForPutBlock;
- buffersForPutBlock = null;
+ byteBufferList = flush.consumeBuffersForPutBlock();
Objects.requireNonNull(byteBufferList, "byteBufferList == null");
} else {
// no data was written since the most recent putBlock
byteBufferList = null;
}
- waitFuturesComplete();
+ waitFuturesComplete(flush);
Review Comment:
Sorry just realized we have an regression here:
```diff
public void executePutBlock(boolean sync, boolean close) throws
IOException {
checkOpen();
long flushPos = totalDataFlushedLength;
- final PendingFlush flush = pendingFlush;
- pendingFlush = new PendingFlush();
final List<StreamBuffer> byteBufferList;
- if (flush.hasBuffersForPutBlock()) {
+ if (pendingFlush.hasBuffersForPutBlock()) {
Objects.requireNonNull(bufferList, "bufferList == null");
- byteBufferList = flush.consumeBuffersForPutBlock();
+ byteBufferList = pendingFlush.consumeBuffersForPutBlock();
Objects.requireNonNull(byteBufferList, "byteBufferList == null");
} else {
// no data was written since the most recent putBlock
byteBufferList = null;
}
- waitFuturesComplete(flush);
+ waitFuturesComplete(pendingFlush);
+ pendingFlush = new PendingFlush();
if (close && putBlockWithoutRaft) {
```
We need to keep the pendingFlush if any of the chunk writes fails.
In current code, if a chunk at the flush boundary failed,
waitFuturesComplete(flush) throws, and the failed futures are dropped along
with flush. A later flush() waits on the fresh, empty pendingFlush and returns
normally. flush() doesn't call checkOpen(), so nothing catches the failure. On
head branch the failed future stays in futures, and flush() find it and throws
again.
You can add a new test to catch it:
```
void flushAfterFailedPutBlockBoundaryStillThrows() throws Exception {
MockDatanodePipeline pipeline = new MockDatanodePipeline();
pipeline.failChunkAfter(3, () -> new IOException("injected"));
BlockDataStreamOutput stream = createStream(pipeline);
byte[] data = randomBytes(400);
assertThrows(IOException.class, () -> stream.write(ByteBuffer.wrap(data),
0, data.length));
assertThrows(IOException.class, stream::flush);
}
```
--
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]
