HDDS-15065. Replace Ratis Snapshot Trigger with DB Flush for Periodic Flush Operations

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -584,15 +584,6 @@ public final class ScmConfigKeys {
   public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT =
           1000L;
 
-  /**
-   * the config will transfer value to ratis config
-   * raft.server.snapshot.creation.gap, used by ratis to take snapshot
-   * when manual trigger using api.
-   */
-  public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_GAP
-      = "ozone.scm.ha.ratis.server.snapshot.creation.gap";
-  public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT =
-      1024L;
   public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_DIR =
           "ozone.scm.ha.ratis.snapshot.dir";
 

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4168,12 +4168,6 @@
       topology cluster tree from SCM.
     </description>
   </property>
-  <property>
-    <name>ozone.scm.ha.ratis.server.snapshot.creation.gap</name>
-    <value>1024</value>
-    <tag>SCM, OZONE</tag>
-    <description>Raft snapshot gap index after which snapshot can be taken.</description>
-  </property>
   <property>
     <name>ozone.scm.ha.dbtransactionbuffer.flush.interval</name>
     <value>60s</value>

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

@@ -231,9 +231,6 @@ private static void setRaftSnapshotProperties(
     Snapshot.setAutoTriggerThreshold(properties,
         ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD,
                 ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT));
-    Snapshot.setCreationGap(properties,
-        ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP,
-            ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT));
   }
 
   public static void checkRatisException(IOException e, String port,

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

@@ -44,7 +44,14 @@ public interface SCMHADBTransactionBuffer
 
   void flush() throws RocksDatabaseException, CodecException;
 
+  void flushIfNeeded(long snapshotWaitTime)
+      throws RocksDatabaseException, CodecException;
+
   boolean shouldFlush(long snapshotWaitTime);
 
   void init() throws RocksDatabaseException, CodecException;
+
+  void beginApplyingTransaction();
+
+  void endApplyingTransaction();
 }

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

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
 
 import com.google.common.base.Preconditions;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -53,6 +54,7 @@ public class SCMHADBTransactionBufferImpl implements SCMHADBTransactionBuffer {
   private final AtomicReference<SnapshotInfo> latestSnapshot
       = new AtomicReference<>();
   private final AtomicLong txFlushPending = new AtomicLong(0);
+  private final AtomicInteger applyingTransactions = new AtomicInteger(0);
   private long lastSnapshotTimeMs = 0;
   private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
 
@@ -124,30 +126,64 @@ public AtomicReference<SnapshotInfo> getLatestSnapshotRef() {
   public void flush() throws RocksDatabaseException, CodecException {
     rwLock.writeLock().lock();
     try {
-      // write latest trx info into trx table in the same batch
-      Table<String, TransactionInfo> transactionInfoTable
-          = metadataStore.getTransactionInfoTable();
-      transactionInfoTable.putWithBatch(currentBatchOperation,
-          TRANSACTION_INFO_KEY, latestTrxInfo);
+      flushUnderWriteLock();
+    } finally {
+      rwLock.writeLock().unlock();
+    }
+  }
 
-      metadataStore.getStore().commitBatchOperation(currentBatchOperation);
-      currentBatchOperation.close();
-      this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
-      // reset batch operation
-      currentBatchOperation = metadataStore.getStore().initBatchOperation();
-
-      DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
-          .getDeletedBlockLog();
-      Preconditions.checkArgument(
-          deletedBlockLog instanceof DeletedBlockLogImpl);
-      ((DeletedBlockLogImpl) deletedBlockLog).onFlush();
+  @Override
+  public void flushIfNeeded(long snapshotWaitTime)
+      throws RocksDatabaseException, CodecException {
+    rwLock.writeLock().lock();
+    try {
+      if (applyingTransactions.get() > 0) {
+        return;
+      }
+      long timeDiff = scm.getSystemClock().millis() - lastSnapshotTimeMs;
+      if (txFlushPending.get() > 0 && timeDiff > snapshotWaitTime) {
+        LOG.debug("Running TransactionFlushTask");
+        flushUnderWriteLock();
+      }
     } finally {
-      txFlushPending.set(0);
-      lastSnapshotTimeMs = scm.getSystemClock().millis();
       rwLock.writeLock().unlock();
     }
   }
 
+  private void flushUnderWriteLock()
+      throws RocksDatabaseException, CodecException {
+    // write latest trx info into trx table in the same batch
+    Table<String, TransactionInfo> transactionInfoTable
+        = metadataStore.getTransactionInfoTable();
+    transactionInfoTable.putWithBatch(currentBatchOperation,
+        TRANSACTION_INFO_KEY, latestTrxInfo);
+
+    metadataStore.getStore().commitBatchOperation(currentBatchOperation);
+    currentBatchOperation.close();
+    this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
+    // reset batch operation
+    currentBatchOperation = metadataStore.getStore().initBatchOperation();
+
+    DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
+        .getDeletedBlockLog();
+    Preconditions.checkArgument(
+        deletedBlockLog instanceof DeletedBlockLogImpl);
+    ((DeletedBlockLogImpl) deletedBlockLog).onFlush();
+
+    txFlushPending.set(0);
+    lastSnapshotTimeMs = scm.getSystemClock().millis();
+  }
+
+  @Override
+  public void beginApplyingTransaction() {
+    applyingTransactions.incrementAndGet();
+  }
+
+  @Override
+  public void endApplyingTransaction() {
+    applyingTransactions.decrementAndGet();
+  }
+
   @Override
   public void init() throws RocksDatabaseException, CodecException {
     metadataStore = scm.getScmMetadataStore();

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

@@ -101,6 +101,19 @@ public AtomicReference<SnapshotInfo> getLatestSnapshotRef() {
     return null;
   }
 
+  @Override
+  public void flushIfNeeded(long snapshotWaitTime) throws RocksDatabaseException {
+    flush();
+  }
+
+  @Override
+  public void beginApplyingTransaction() {
+  }
+
+  @Override
+  public void endApplyingTransaction() {
+  }
+
   @Override
   public void flush() throws RocksDatabaseException {
     rwLock.writeLock().lock();

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

@@ -142,8 +142,7 @@ private void createStartTransactionBufferMonitor() {
         OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL_DEFAULT,
         TimeUnit.MILLISECONDS);
     SCMHATransactionBufferMonitorTask monitorTask
-        = new SCMHATransactionBufferMonitorTask(
-        transactionBuffer, ratisServer, interval);
+        = new SCMHATransactionBufferMonitorTask(transactionBuffer, interval);
     trxBufferMonitorService =
         new BackgroundSCMService.Builder().setClock(scm.getSystemClock())
             .setScmContext(scm.getScmContext())

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

@@ -18,7 +18,6 @@
 package org.apache.hadoop.hdds.scm.ha;
 
 import java.io.IOException;
-import org.apache.ratis.statemachine.SnapshotInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,39 +28,24 @@
 public class SCMHATransactionBufferMonitorTask implements Runnable {
   private static final Logger LOG =
       LoggerFactory.getLogger(SCMHATransactionBufferMonitorTask.class);
-  private final SCMRatisServer server;
   private final SCMHADBTransactionBuffer transactionBuffer;
   private final long flushInterval;
 
   /**
    * SCMService related variables.
    */
   public SCMHATransactionBufferMonitorTask(
-      SCMHADBTransactionBuffer transactionBuffer,
-      SCMRatisServer server, long flushInterval) {
+      SCMHADBTransactionBuffer transactionBuffer, long flushInterval) {
     this.flushInterval = flushInterval;
     this.transactionBuffer = transactionBuffer;
-    this.server = server;
   }
 
   @Override
   public void run() {
-    if (transactionBuffer.shouldFlush(flushInterval)) {
-      LOG.debug("Running TransactionFlushTask");
-      // set latest snapshot to null for force snapshot
-      // the value will be reset again when snapshot is taken
-      final SnapshotInfo lastSnapshot = transactionBuffer
-          .getLatestSnapshotRef().getAndSet(null);
-      try {
-        server.triggerSnapshot();
-      } catch (IOException e) {
-        LOG.error("Snapshot request is failed", e);
-      } finally {
-        // under failure case, if unable to take snapshot, its value
-        // is reset to previous known value
-        transactionBuffer.getLatestSnapshotRef().compareAndSet(
-            null, lastSnapshot);
-      }
+    try {
+      transactionBuffer.flushIfNeeded(flushInterval);
+    } catch (IOException e) {
+      LOG.error("TransactionFlushTask is failed", e);
     }
   }
 }

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

@@ -150,6 +150,7 @@ public CompletableFuture<Message> applyTransaction(
       final TransactionContext trx) {
     final CompletableFuture<Message> applyTransactionFuture =
         new CompletableFuture<>();
+    transactionBuffer.beginApplyingTransaction();
     try {
       final SCMRatisRequest request = SCMRatisRequest.decode(
           Message.valueOf(trx.getStateMachineLogEntry().getLogData()));
@@ -182,6 +183,8 @@ public CompletableFuture<Message> applyTransaction(
     } catch (Exception ex) {
       applyTransactionFuture.completeExceptionally(ex);
       ExitUtils.terminate(1, ex.getMessage(), ex, StateMachine.LOG);
+    } finally {
+      transactionBuffer.endApplyingTransaction();
     }
     return applyTransactionFuture;
   }
@@ -343,23 +346,28 @@ public long takeSnapshot() throws IOException {
       return lastAppliedIndex;
     }
 
-    long startTime = Time.monotonicNow();
+    transactionBuffer.beginApplyingTransaction();
+    try {
+      long startTime = Time.monotonicNow();
 
-    TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
-    final TransactionInfo lastAppliedTrxInfo = TransactionInfo.valueOf(lastTermIndex);
+      TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
+      final TransactionInfo lastAppliedTrxInfo = TransactionInfo.valueOf(lastTermIndex);
 
-    if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
-      transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
-      transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
-    } else {
-      lastAppliedIndex = latestTrxInfo.getTransactionIndex();
-    }
+      if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
+        transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
+        transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
+      } else {
+        lastAppliedIndex = latestTrxInfo.getTransactionIndex();
+      }
 
-    transactionBuffer.flush();
+      transactionBuffer.flush();
 
-    LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
-        lastAppliedIndex, Time.monotonicNow() - startTime);
-    return lastAppliedIndex;
+      LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
+          lastAppliedIndex, Time.monotonicNow() - startTime);
+      return lastAppliedIndex;
+    } finally {
+      transactionBuffer.endApplyingTransaction();
+    }
   }
 
   @Override
@@ -379,7 +387,12 @@ public void notifyTermIndexUpdated(long term, long index) {
     }
 
     if (transactionBuffer != null) {
-      transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index));
+      transactionBuffer.beginApplyingTransaction();
+      try {
+        transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index));
+      } finally {
+        transactionBuffer.endApplyingTransaction();
+      }
     }
 
     if (currentLeaderTerm.get() == term) {