Data integration tests passing

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/ContainerChecksumTreeManager.java

@@ -89,6 +89,9 @@ public void writeContainerDataTree(ContainerData data, ContainerMerkleTree tree)
     long containerID = data.getContainerID();
     Lock writeLock = getLock(containerID);
     writeLock.lock();
+    // If there is an error generating the tree and we cannot obtain a final checksum, use 0 to indicate a metadata
+    // failure.
+    long dataChecksum = 0;
     try {
       ContainerProtos.ContainerChecksumInfo.Builder checksumInfoBuilder = null;
       try {
@@ -103,14 +106,16 @@ public void writeContainerDataTree(ContainerData data, ContainerMerkleTree tree)
 
       ContainerProtos.ContainerMerkleTree treeProto = captureLatencyNs(metrics.getCreateMerkleTreeLatencyNS(),
           tree::toProto);
-      long dataChecksum = treeProto.getDataChecksum();
-      data.setDataChecksum(dataChecksum);
       checksumInfoBuilder
           .setContainerID(containerID)
           .setContainerMerkleTree(treeProto);
       write(data, checksumInfoBuilder.build());
+      // If write succeeds, update the checksum in memory. Otherwise 0 will be used to indicate the metadata failure.
+      dataChecksum = treeProto.getDataChecksum();
       LOG.debug("Data merkle tree for container {} updated with container checksum {}", containerID, dataChecksum);
     } finally {
+      // Even if persisting the tree fails, we should still update the data checksum in memory to report back to SCM.
+      data.setDataChecksum(dataChecksum);
       writeLock.unlock();
     }
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/BackgroundContainerDataScanner.java

@@ -92,6 +92,13 @@ public void scanContainer(Container<?> c)
     if (result.isDeleted()) {
       LOG.debug("Container [{}] has been deleted during the data scan.", containerId);
     } else {
+      // Merkle tree write failure should not abort the scanning process. Continue marking the scan as completed.
+      try {
+        checksumManager.writeContainerDataTree(containerData, result.getDataTree());
+      } catch (IOException ex) {
+        LOG.error("Failed to write container merkle tree for container {}", containerId, ex);
+      }
+
       if (!result.isHealthy()) {
         logUnhealthyScanResult(containerId, result, LOG);
 
@@ -104,7 +111,6 @@ public void scanContainer(Container<?> c)
         }
       }
       metrics.incNumContainersScanned();
-      checksumManager.writeContainerDataTree(containerData, result.getDataTree());
     }
 
     Instant now = Instant.now();

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OnDemandContainerDataScanner.java

@@ -141,6 +141,12 @@ private static void performOnDemandScan(Container<?> container) {
       if (result.isDeleted()) {
         LOG.debug("Container [{}] has been deleted during the data scan.", containerId);
       } else {
+        // Merkle tree write failure should not abort the scanning process. Continue marking the scan as completed.
+        try {
+          instance.checksumManager.writeContainerDataTree(containerData, result.getDataTree());
+        } catch (IOException ex) {
+          LOG.error("Failed to write container merkle tree for container {}", containerId, ex);
+        }
         if (!result.isHealthy()) {
           logUnhealthyScanResult(containerId, result, LOG);
           boolean containerMarkedUnhealthy = instance.containerController
@@ -150,7 +156,6 @@ private static void performOnDemandScan(Container<?> container) {
           }
         }
         instance.metrics.incNumContainersScanned();
-        instance.checksumManager.writeContainerDataTree(containerData, result.getDataTree());
       }
 
       Instant now = Instant.now();

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scanner/TestBackgroundContainerMetadataScannerIntegration.java

@@ -98,14 +98,14 @@ void testCorruptionDetected(TestContainerCorruptions corruption)
     assertEquals(State.CLOSED, closedContainer.getContainerState());
     assertTrue(containerChecksumFileExists(closedContainerID));
     waitForScmToSeeReplicaState(closedContainerID, CLOSED);
-    long initialReportedClosedChecksum = getContainerReplica(closedContainerID).getDataChecksum();
+    long initialClosedChecksum = getContainerReplica(closedContainerID).getDataChecksum();
+    assertNotEquals(0, initialClosedChecksum);
 
     long openContainerID = writeDataToOpenContainer();
     Container<?> openContainer = getDnContainer(openContainerID);
     assertEquals(State.OPEN, openContainer.getContainerState());
-    long initialReportedOpenChecksum = getContainerReplica(openContainerID).getDataChecksum();
     // Open containers should not yet have a checksum generated.
-    assertEquals(0, initialReportedOpenChecksum);
+    assertEquals(0, getContainerReplica(openContainerID).getDataChecksum());
 
     // Corrupt both containers.
     corruption.applyTo(closedContainer);
@@ -120,10 +120,12 @@ void testCorruptionDetected(TestContainerCorruptions corruption)
         500, 5000);
 
     // Wait for SCM to get reports of the unhealthy replicas.
+    // The metadata scanner does not generate data checksums and the other scanners have been turned off for this
+    // test, so the data checksums should not change.
     waitForScmToSeeReplicaState(closedContainerID, UNHEALTHY);
-    assertNotEquals(initialReportedClosedChecksum, getContainerReplica(closedContainerID).getDataChecksum());
+    assertEquals(initialClosedChecksum, getContainerReplica(closedContainerID).getDataChecksum());
     waitForScmToSeeReplicaState(openContainerID, UNHEALTHY);
-    assertNotEquals(initialReportedOpenChecksum, getContainerReplica(openContainerID).getDataChecksum());
+    assertEquals(0, getContainerReplica(openContainerID).getDataChecksum());
 
     // Once the unhealthy replica is reported, the open container's lifecycle
     // state in SCM should move to closed.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scanner/TestOnDemandContainerDataScannerIntegration.java

@@ -34,6 +34,7 @@
 
 import java.util.Collection;
 
+import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State.CLOSED;
 import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State.UNHEALTHY;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -102,9 +103,11 @@ void testCorruptionDetected(TestContainerCorruptions corruption)
     // Container corruption has not yet been introduced.
     Container<?> container = getDnContainer(containerID);
     assertEquals(State.CLOSED, container.getContainerState());
-    long initialReportedDataChecksum = getContainerReplica(containerID).getDataChecksum();
     assertTrue(containerChecksumFileExists(containerID));
 
+    waitForScmToSeeReplicaState(containerID, CLOSED);
+    long initialReportedDataChecksum = getContainerReplica(containerID).getDataChecksum();
+
     // Corrupt the container.
     corruption.applyTo(container);
     // This method will check that reading from the corrupted key returns an