diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java index 06a73b33373..739858d5de4 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java @@ -119,6 +119,15 @@ public class KeyValueContainer implements Container { // container are synchronous. private Set pendingPutBlockCache; + // Set of Blocks (LocalIDs) that have already been finalized by a PutBlock + // with the end-of-block (eof) flag set. PutBlock is not idempotent, so there + // should be only one such call per block. It is used to detect and + // ignore any further writes on a block after its final PutBlock, (see HDDS-12007). + // Like pendingPutBlockCache, it is only populated for OPEN or CLOSING containers and + // cleared when the container is closed. Writes to the container are synchronous, so no + // explicit synchronization is required. + private Set eofBlockCache; + private boolean bCheckChunksFilePath; private static FaultInjector faultInjector; @@ -130,10 +139,12 @@ public KeyValueContainer(KeyValueContainerData containerData, this.containerData = containerData; if (this.containerData.isOpen() || this.containerData.isClosing()) { // If container is not in OPEN or CLOSING state, there cannot be block - // writes to the container. So pendingPutBlockCache is not needed. + // writes to the container. So pendingPutBlockCache and eofBlockCache are not needed. this.pendingPutBlockCache = new HashSet<>(); + this.eofBlockCache = new HashSet<>(); } else { this.pendingPutBlockCache = Collections.emptySet(); + this.eofBlockCache = Collections.emptySet(); } DatanodeConfiguration dnConf = config.getObject(DatanodeConfiguration.class); @@ -403,6 +414,7 @@ public void markContainerUnhealthy() throws StorageContainerException { updateContainerState(UNHEALTHY); } clearPendingPutBlockCache(); + clearEofBlockCache(); } finally { writeUnlock(); } @@ -470,6 +482,7 @@ private void closeAndFlushIfNeeded(Runnable closer) flushAndSyncDB(); updateContainerData(closer); clearPendingPutBlockCache(); + clearEofBlockCache(); } finally { writeUnlock(); } @@ -861,6 +874,34 @@ public void removeFromPendingPutBlockCache(long localID) { pendingPutBlockCache.remove(localID); } + /** + * Return whether the given localID of a block has already been finalized by + * a PutBlock with the end-of-block flag set. + */ + public boolean isBlockFinalizedByEof(long localID) { + return eofBlockCache.contains(localID); + } + + /** + * Record that the given localID of a block has been finalized by a PutBlock + * with the end-of-block flag set, so that subsequent writes on it can be + * ignored. + */ + public void addToEofBlockCache(long localID) + throws StorageContainerException { + try { + eofBlockCache.add(localID); + } catch (UnsupportedOperationException e) { + // Getting an UnsupportedOperationException here implies that the + // eofBlockCache is an Empty Set. This should not happen if the + // container is in OPEN or CLOSING state. Log the exception here and + // throw a non-Runtime exception so that putBlock request fails. + String msg = "Failed to add block " + localID + " to eofBlockCache for " + containerData; + LOG.error(msg, e); + throw new StorageContainerException(msg, CONTAINER_INTERNAL_ERROR); + } + } + /** * When a container is closed, quasi-closed or marked unhealthy, clear the * pendingPutBlockCache as there won't be any more writes to the container. @@ -870,6 +911,15 @@ private void clearPendingPutBlockCache() { pendingPutBlockCache = Collections.emptySet(); } + /** + * When a container is closed, quasi-closed or marked unhealthy, clear the + * eofBlockCache as there won't be any more writes to the container. + */ + private void clearEofBlockCache() { + eofBlockCache.clear(); + eofBlockCache = Collections.emptySet(); + } + /** * Returns KeyValueContainerReport for the KeyValueContainer. */ 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 46929633d09..5ce7afa501f 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 @@ -205,10 +205,17 @@ public long persistPutBlock(KeyValueContainer container, return data.getSize(); } - // Check if the block is present in the pendingPutBlockCache for the - // container to determine whether the blockCount is already incremented - // for this block in the DB or not. long localID = data.getLocalID(); + + // PutBlock is not idempotent; ignore duplicate eof writes (HDDS-12007). + if (container.isBlockFinalizedByEof(localID)) { + LOG.warn("Ignoring write on block {} which has already been finalized " + + "by a PutBlock with the end-of-block flag set. PutBlock is not " + + "idempotent", + data.getBlockID()); + return data.getSize(); + } + // 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 @@ -223,6 +230,10 @@ public long persistPutBlock(KeyValueContainer container, data.setBlockCommitSequenceId(bcsId); } } + + // Check if the block is present in the pendingPutBlockCache for the + // container to determine whether the blockCount is already incremented + // for this block in the DB or not. boolean isBlockInCache = container.isBlockInPendingPutBlockCache(localID); boolean incrBlockCount = false; @@ -294,6 +305,13 @@ public long persistPutBlock(KeyValueContainer container, container.removeFromPendingPutBlockCache(localID); } + // Track the block as finalized so that any subsequent write on it can be + // detected and ignored, since PutBlock is not idempotent. Only OPEN and + // CLOSING containers maintain eofBlockCache. + if (endOfBlock && (containerData.isOpen() || containerData.isClosing())) { + container.addToEofBlockCache(localID); + } + if (LOG.isDebugEnabled()) { LOG.debug( "Block " + data.getBlockID() + " successfully committed with bcsId " diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java index eb58ac3ce84..74181fa8dbe 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java @@ -21,6 +21,7 @@ import static org.apache.hadoop.ozone.OzoneConsts.INCREMENTAL_CHUNK_LIST; import static org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil.isSameSchemaVersion; import static org.apache.hadoop.ozone.container.keyvalue.impl.BlockManagerImpl.FULL_CHUNK; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -53,6 +54,7 @@ import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer; import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils; +import org.apache.ozone.test.GenericTestUtils.LogCapturer; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.io.TempDir; @@ -468,4 +470,64 @@ public void testFlush3(ContainerTestVersionInfo versionInfo) assertEquals(chunkLimit * i, chunkInfos.get(i).getOffset()); } } + + @ContainerTestVersionInfo.ContainerTest + public void testPutBlockAfterEndOfBlockIgnored( + ContainerTestVersionInfo versionInfo) throws Exception { + initTest(versionInfo); + // Finalize the block with an end-of-block PutBlock. + blockManager.putBlock(keyValueContainer, blockData, true); + assertEquals(1, keyValueContainer.getContainerData().getBlockCount()); + long finalizedSize = blockManager.getBlock(keyValueContainer, + blockData.getBlockID()).getSize(); + + // A subsequent PutBlock on the same block is ignored (no exception), + // since PutBlock is not idempotent and the block is already finalized. + // Uses different content to verify that the finalized block is not overwritten. + BlockData duplicate = new BlockData(blockData.getBlockID()); + List chunkList = new ArrayList<>(); + chunkList.add(new ChunkInfo("data", 0, 4096).getProtoBufMessage()); + duplicate.setChunks(chunkList); + LogCapturer logCapturer = LogCapturer.captureLogs(BlockManagerImpl.class); + blockManager.putBlock(keyValueContainer, duplicate, true); + + // A warning is logged when the write is ignored. + assertThat(logCapturer.getOutput()).contains("has already been finalized"); + + // Block count is unchanged and the finalized block data is untouched. + assertEquals(1, keyValueContainer.getContainerData().getBlockCount()); + assertEquals(finalizedSize, blockManager.getBlock(keyValueContainer, + blockData.getBlockID()).getSize()); + } + + @ContainerTestVersionInfo.ContainerTest + public void testWriteAfterEndOfBlockIgnoredIncremental( + ContainerTestVersionInfo versionInfo) throws Exception { + initTest(versionInfo); + Assumptions.assumeFalse( + isSameSchemaVersion(schemaVersion, OzoneConsts.SCHEMA_V1)); + long containerID = 1; + long blockNo = 2; + // incremental write, block not yet finalized + BlockData first = createBlockData(containerID, blockNo, 1, 0, 1024, 1); + blockManager.putBlock(keyValueContainer, first, false); + // end-of-block PutBlock finalizes the block + BlockData eof = createBlockData(containerID, blockNo, 1, 0, 2048, 2); + blockManager.putBlock(keyValueContainer, eof, true); + BlockData finalized = blockManager.getBlock(keyValueContainer, + new BlockID(containerID, blockNo)); + long finalizedSize = finalized.getSize(); + long finalizedBcsId = finalized.getBlockCommitSequenceId(); + + // A further write on the same block (with a higher bcsId) is ignored, + // leaving the finalized block untouched. + BlockData afterEof = createBlockData(containerID, blockNo, 1, 0, 3072, 3); + blockManager.putBlock(keyValueContainer, afterEof, true); + + BlockData afterIgnore = blockManager.getBlock(keyValueContainer, + new BlockID(containerID, blockNo)); + assertEquals(finalizedSize, afterIgnore.getSize()); + assertEquals(finalizedBcsId, afterIgnore.getBlockCommitSequenceId()); + assertEquals(1, keyValueContainer.getContainerData().getBlockCount()); + } }