HDDS-14572. ReInit DeletedBlocksTransactionSummary after SCM leader transfer

Author: ChenSammi

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMDeletedBlockTransactionStatusManager.java

@@ -424,6 +424,14 @@ public void clear() {
     scmDeleteBlocksCommandStatusManager.clear();
     transactionToDNsCommitMap.clear();
     txSizeMap.clear();
+    try {
+      initDataDistributionData();
+    } catch (IOException e) {
+      LOG.warn("Failed to initialize Storage space distribution data. The feature will continue with current " +
+              "totalBlockCount {}, totalBlockCount {}, totalBlocksSize {} and totalReplicatedBlocksSize {}. " +
+              "There is a high chance that the real data and current data has a fixed gap.",
+          totalBlockCount.get(), totalBlocksSize.get(), totalBlocksSize.get(), totalReplicatedBlocksSize.get());
+    }
   }
 
   public void cleanAllTimeoutSCMCommand(long timeoutMs) {
@@ -672,8 +680,9 @@ private void initDataDistributionData() throws IOException {
       totalBlockCount.set(summary.getTotalBlockCount());
       totalBlocksSize.set(summary.getTotalBlockSize());
       totalReplicatedBlocksSize.set(summary.getTotalBlockReplicatedSize());
-      LOG.info("Data distribution is enabled with totalBlockCount {} totalBlocksSize {}",
-          totalBlockCount.get(), totalBlocksSize.get());
+      LOG.info("Storage space distribution is initialized with totalTxCount {}, totalBlockCount {}, " +
+              "totalBlocksSize {} and totalReplicatedBlocksSize {}", totalTxCount.get(),
+          totalBlockCount.get(), totalBlocksSize.get(), totalReplicatedBlocksSize.get());
     }
   }
 
@@ -688,8 +697,7 @@ private DeletedBlocksTransactionSummary loadDeletedBlocksSummary() throws IOExce
       }
       return DeletedBlocksTransactionSummary.parseFrom(byteString);
     } catch (IOException e) {
-      LOG.error("Failed to get property {} for service {}. DataDistribution function will be disabled.",
-          propertyName, SERVICE_NAME, e);
+      LOG.error("Failed to get property {} for service {}.", propertyName, SERVICE_NAME, e);
       throw new IOException("Failed to get property " + propertyName, e);
     }
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java

@@ -296,6 +296,12 @@ public void notifyLeaderChanged(RaftGroupMemberId groupMemberId,
         currentLeaderTerm.get());
     scm.getSequenceIdGen().invalidateBatch();
 
+    try {
+      transactionBuffer.flush();
+    } catch (Exception ex) {
+      ExitUtils.terminate(1, "Failed to flush transactionBuffer", ex, StateMachine.LOG);
+    }
+
     DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
         .getDeletedBlockLog();
     Preconditions.checkArgument(

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/upgrade/TestScmDataDistributionFinalization.java

@@ -18,6 +18,11 @@
 package org.apache.hadoop.hdds.upgrade;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_COMMAND_STATUS_REPORT_INTERVAL;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_REPORT_INTERVAL;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT;
 import static org.apache.hadoop.hdds.client.ReplicationFactor.THREE;
 import static org.apache.hadoop.hdds.client.ReplicationType.RATIS;
@@ -109,12 +114,14 @@ public void init(OzoneConfiguration conf,
     configurator.setUpgradeFinalizationExecutor(executor);
 
     conf.setInt(SCMStorageConfig.TESTING_INIT_LAYOUT_VERSION_KEY, HDDSLayoutFeature.HBASE_SUPPORT.layoutVersion());
-    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
-        TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
-        TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
-        100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(HDDS_COMMAND_STATUS_REPORT_INTERVAL, 100, MILLISECONDS);
+    conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(HDDS_PIPELINE_REPORT_INTERVAL, 100, TimeUnit.MILLISECONDS);
+
     ScmConfig scmConfig = conf.getObject(ScmConfig.class);
     scmConfig.setBlockDeletionInterval(Duration.ofMillis(100));
     conf.setFromObject(scmConfig);
@@ -123,6 +130,7 @@ public void init(OzoneConfiguration conf,
     DatanodeConfiguration dnConf =
         conf.getObject(DatanodeConfiguration.class);
     dnConf.setBlockDeletionInterval(Duration.ofMillis(100));
+    dnConf.setBlockDeleteCommandWorkerInterval(Duration.ofMillis(100));
     conf.setFromObject(dnConf);
 
     MiniOzoneHAClusterImpl.Builder clusterBuilder = MiniOzoneCluster.newHABuilder(conf);
@@ -344,13 +352,39 @@ public void testFinalizationNonEmptyClusterDataDistribution() throws Exception {
     assertEquals(value.getBytes(UTF_8).length, summary.getTotalBlockSize());
     assertEquals(value.getBytes(UTF_8).length * 3, summary.getTotalBlockReplicatedSize());
 
+    // transfer SCM leader
+    String newLeaderScmId = null;
+    for (StorageContainerManager scm: cluster.getStorageContainerManagersList()) {
+      if (scm != activeSCM) {
+        newLeaderScmId = scm.getScmId();
+        break;
+      }
+    }
+    cluster.getStorageContainerLocationClient().transferLeadership(newLeaderScmId);
+    StorageContainerManager newActiveSCM = cluster.getActiveSCM();
+    deletedBlockLog = (DeletedBlockLogImpl) newActiveSCM.getScmBlockManager().getDeletedBlockLog();
+    SCMDeletedBlockTransactionStatusManager newStatusManager =
+        deletedBlockLog.getSCMDeletedBlockTransactionStatusManager();
+    // new leader SCM should have the right deletion tx summary
+    summary = newStatusManager.getTransactionSummary();
+    assertEquals(1, summary.getTotalTransactionCount());
+    assertEquals(1, summary.getTotalBlockCount());
+    assertEquals(value.getBytes(UTF_8).length, summary.getTotalBlockSize());
+    assertEquals(value.getBytes(UTF_8).length * 3, summary.getTotalBlockReplicatedSize());
+
+    // flush buffer and start SCMBlockDeletingService
+    for (StorageContainerManager scm: cluster.getStorageContainerManagersList()) {
+      flushDBTransactionBuffer(scm);
+      scm.getScmBlockManager().getSCMBlockDeletingService().start();
+    }
+
     // force close the container so that block can be deleted
-    activeSCM.getClientProtocolServer().closeContainer(
+    newActiveSCM.getClientProtocolServer().closeContainer(
         keyDetails.getOzoneKeyLocations().get(0).getContainerID());
     // wait for container to be closed
     GenericTestUtils.waitFor(() -> {
       try {
-        return activeSCM.getClientProtocolServer().getContainer(
+        return newActiveSCM.getClientProtocolServer().getContainer(
             keyDetails.getOzoneKeyLocations().get(0).getContainerID())
             .getState() == HddsProtos.LifeCycleState.CLOSED;
       } catch (IOException e) {
@@ -359,15 +393,15 @@ public void testFinalizationNonEmptyClusterDataDistribution() throws Exception {
       }
     }, 100, 5000);
 
-    // flush buffer and start SCMBlockDeletingService
-    for (StorageContainerManager scm: cluster.getStorageContainerManagersList()) {
-      flushDBTransactionBuffer(scm);
-      scm.getScmBlockManager().getSCMBlockDeletingService().start();
-    }
-
     // wait for block deletion transactions to be confirmed by DN
     GenericTestUtils.waitFor(
-        () -> statusManager.getTransactionSummary().getTotalTransactionCount() == 0, 100, 30000);
+        () -> newStatusManager.getTransactionSummary().getTotalTransactionCount() == 0, 100, 30000);
+
+    // transfer leader back to old SCM and verify
+    cluster.getStorageContainerLocationClient().transferLeadership(activeSCM.getScmId());
+    deletedBlockLog = (DeletedBlockLogImpl) activeSCM.getScmBlockManager().getDeletedBlockLog();
+    summary = deletedBlockLog.getSCMDeletedBlockTransactionStatusManager().getTransactionSummary();
+    assertEquals(EMPTY_SUMMARY, summary);
   }
 
   private Map<Long, List<DeletedBlock>> generateDeletedBlocks(int dataSize, boolean withSize) {