HDDS-15065. Handling partial flush scenarios

Author: priyeshkaratha

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/SCMHATransactionBufferMonitorTask.java

@@ -42,13 +42,10 @@ public SCMHATransactionBufferMonitorTask(
 
   @Override
   public void run() {
-    if (transactionBuffer.shouldFlush(flushInterval)) {
-      LOG.debug("Running TransactionFlushTask");
-      try {
-        transactionBuffer.flush();
-      } catch (IOException e) {
-        LOG.error("TransactionFlushTask is failed", e);
-      }
+    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;
   }