HDDS-15723. Ignore writes after PutBlock with end-of-block flag

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -119,6 +119,15 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
   // container are synchronous.
   private Set<Long> 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<Long> 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.
    */

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 "

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<ContainerProtos.ChunkInfo> 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());
+  }
 }