hanishakoneru commented on a change in pull request #3034:
URL: https://github.com/apache/ozone/pull/3034#discussion_r801066168
##########
File path:
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
##########
@@ -415,41 +422,54 @@ void updateCommitInfo(XceiverClientReply reply,
List<ChunkBuffer> buffers) {
ContainerCommandResponseProto> flushFuture = null;
try {
BlockData blockData = containerBlockData.build();
- XceiverClientReply asyncReply =
- putBlockAsync(xceiverClient, blockData, close, token);
+ boolean firstBlockPutCall = updateBlockCount.getAndSet(false);
+ XceiverClientReply asyncReply = putBlockAsync(xceiverClient, blockData,
+ firstBlockPutCall, close, token);
CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future =
asyncReply.getResponse();
- flushFuture = future.thenApplyAsync(e -> {
+ flushFuture = future.thenApplyAsync(response -> {
try {
- validateResponse(e);
+ validateResponse(response);
} catch (IOException sce) {
throw new CompletionException(sce);
}
// if the ioException is not set, putBlock is successful
- if (getIoException() == null && !force) {
- BlockID responseBlockID = BlockID.getFromProtobuf(
- e.getPutBlock().getCommittedBlockLength().getBlockID());
- Preconditions.checkState(blockID.get().getContainerBlockID()
- .equals(responseBlockID.getContainerBlockID()));
- // updates the bcsId of the block
- blockID.set(responseBlockID);
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Adding index " + asyncReply.getLogIndex() + " flushLength "
- + flushPos + " numBuffers " + byteBufferList.size()
- + " blockID " + blockID + " bufferPool size" + bufferPool
- .getSize() + " currentBufferIndex " + bufferPool
- .getCurrentBufferIndex());
+ if (getIoException() == null) {
+ if (!force) {
+ BlockID responseBlockID = BlockID.getFromProtobuf(
+ response.getPutBlock().getCommittedBlockLength().getBlockID());
+ Preconditions.checkState(blockID.get().getContainerBlockID()
+ .equals(responseBlockID.getContainerBlockID()));
+ // updates the bcsId of the block
+ blockID.set(responseBlockID);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Adding index " + asyncReply.getLogIndex() + " flushLength "
+ + flushPos + " numBuffers " + byteBufferList.size()
+ + " blockID " + blockID + " bufferPool size" + bufferPool
+ .getSize() + " currentBufferIndex " + bufferPool
+ .getCurrentBufferIndex());
+ }
+ // for standalone protocol, logIndex will always be 0.
+ updateCommitInfo(asyncReply, byteBufferList);
+ }
+ } else {
+ // PutBlock not successful -> reset updateBlockCount if it was set
+ // to true
+ if (firstBlockPutCall) {
+ updateBlockCount.compareAndSet(true, false);
Review comment:
Thanks @adoroszlai for catching this.
Yes agree, it should be set to true in case of failure.
--
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]
