jojochuang commented on code in PR #7047:
URL: https://github.com/apache/ozone/pull/7047#discussion_r1722459892
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java:
##########
@@ -406,41 +401,35 @@ private void updatePutBlockLength() {
* @throws IOException if error occurred
*/
- // In this case, the data is already cached in the currentBuffer.
+ // In this case, the data is already cached in the allocated buffers in the
BufferPool.
public synchronized void writeOnRetry(long len) throws IOException {
if (len == 0) {
return;
}
+ List<ChunkBuffer> allocatedBuffers = bufferPool.getAllocatedBuffers();
if (LOG.isDebugEnabled()) {
- LOG.debug("Retrying write length {} for blockID {}", len, blockID);
+ LOG.debug("{}: Retrying write length {} on target blockID {}, {}
buffers", this, len, blockID,
+ allocatedBuffers.size());
}
Preconditions.checkArgument(len <=
streamBufferArgs.getStreamBufferMaxSize());
int count = 0;
- List<ChunkBuffer> allocatedBuffers = bufferPool.getAllocatedBuffers();
while (len > 0) {
ChunkBuffer buffer = allocatedBuffers.get(count);
long writeLen = Math.min(buffer.position(), len);
- if (!buffer.hasRemaining()) {
- writeChunk(buffer);
- }
len -= writeLen;
count++;
writtenDataLength += writeLen;
- // we should not call isBufferFull/shouldFlush here.
- // The buffer might already be full as whole data is already cached in
- // the buffer. We should just validate
- // if we wrote data of size streamBufferMaxSize/streamBufferFlushSize to
- // call for handling full buffer/flush buffer condition.
- if (writtenDataLength % streamBufferArgs.getStreamBufferFlushSize() ==
0) {
- // reset the position to zero as now we will be reading the
- // next buffer in the list
- updateWriteChunkLength();
- updatePutBlockLength();
- CompletableFuture<PutBlockResult> putBlockResultFuture =
executePutBlock(false, false);
- recordWatchForCommitAsync(putBlockResultFuture);
- }
- if (writtenDataLength == streamBufferArgs.getStreamBufferMaxSize()) {
- handleFullBuffer();
+ updateWriteChunkLength();
+ updatePutBlockLength();
+ LOG.debug("Write chunk on retry buffer = {}", buffer);
+ CompletableFuture<PutBlockResult> f = writeChunkAndPutBlock(buffer,
false);
Review Comment:
writeChunkAndPutBlock() was initially used only when WriteChunk/PutBlock
piggybacking is enabled.
This can cause compatibility problems if server side does not support
piggybacking.
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java:
##########
@@ -102,6 +119,37 @@ void checkStream() throws IOException {
}
}
+ /** Register when a call (write or flush) is received on this block. */
+ void registerCallReceived() {
+ inflightCalls.incrementAndGet();
+ }
+
+ /**
+ * Register when a call (write or flush) is finised on this block.
Review Comment:
```suggestion
* Register when a call (write or flush) is finished on this block.
```
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java:
##########
@@ -69,6 +73,18 @@ public class BlockOutputStreamEntry extends OutputStream {
private final StreamBufferArgs streamBufferArgs;
private final Supplier<ExecutorService> executorServiceSupplier;
+ /**
+ * An indicator that this BlockOutputStream is created to handoff writes
from another faulty BlockOutputStream.
+ * Once this flag is on, this BlockOutputStream can only handle
writeOneRetry.
Review Comment:
typo: writeOnRetry
##########
hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java:
##########
@@ -290,8 +325,15 @@ private int writeToOutputStream(BlockOutputStreamEntry
current,
* @param exception actual exception that occurred
* @throws IOException Throws IOException if Write fails
*/
- private synchronized void handleException(BlockOutputStreamEntry streamEntry,
- IOException exception) throws IOException {
+ private void handleExceptionInternal(BlockOutputStreamEntry streamEntry,
IOException exception) throws IOException {
+ try {
+ // Wait for all pending flushes in the faulty stream. It's possible that
a prior write is pending completion
+ // successfully. Errors are ignored here and will be handled by the
individual flush call. We just wall to ensure
Review Comment:
```suggestion
// successfully. Errors are ignored here and will be handled by the
individual flush call. We just want to ensure
```
--
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]
