HDDS-11712. Process other DeletedBlocksTransaction before retrying failed one

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

@@ -93,6 +93,7 @@ public class DeletedBlockLogImpl
   private long scmCommandTimeoutMs = Duration.ofSeconds(300).toMillis();
 
   private static final int LIST_ALL_FAILED_TRANSACTIONS = -1;
+  private long lastProcessedTransactionId;
 
   public DeletedBlockLogImpl(ConfigurationSource conf,
       StorageContainerManager scm,
@@ -115,6 +116,7 @@ public DeletedBlockLogImpl(ConfigurationSource conf,
     this.scmContext = scm.getScmContext();
     this.sequenceIdGen = scm.getSequenceIdGen();
     this.metrics = metrics;
+    this.lastProcessedTransactionId = -1L;
     this.transactionStatusManager =
         new SCMDeletedBlockTransactionStatusManager(deletedBlockLogStateManager,
             containerManager, this.scmContext, metrics, scmCommandTimeoutMs);
@@ -341,9 +343,13 @@ public DatanodeDeletedBlockTransactions getTransactions(
           scmCommandTimeoutMs);
       DatanodeDeletedBlockTransactions transactions =
           new DatanodeDeletedBlockTransactions();
+      long firstProcessedTransactionKey = -1;
       try (TableIterator<Long,
           ? extends Table.KeyValue<Long, DeletedBlocksTransaction>> iter =
                deletedBlockLogStateManager.getReadOnlyIterator()) {
+        if (lastProcessedTransactionId != -1) {
+          iter.seek(lastProcessedTransactionId);
+        }
         // Get the CmdStatus status of the aggregation, so that the current
         // status of the specified transaction can be found faster
         Map<UUID, Map<Long, CmdStatus>> commandStatus =
@@ -360,6 +366,12 @@ public DatanodeDeletedBlockTransactions getTransactions(
             transactions.getBlocksDeleted() < blockDeletionLimit) {
           Table.KeyValue<Long, DeletedBlocksTransaction> keyValue = iter.next();
           DeletedBlocksTransaction txn = keyValue.getValue();
+          if (firstProcessedTransactionKey == keyValue.getKey()) {
+            break;
+          }
+          if (firstProcessedTransactionKey == -1) {
+            firstProcessedTransactionKey = keyValue.getKey();
+          }
           final ContainerID id = ContainerID.valueOf(txn.getContainerID());
           try {
             // HDDS-7126. When container is under replicated, it is possible
@@ -386,6 +398,25 @@ public DatanodeDeletedBlockTransactions getTransactions(
             LOG.warn("Container: {} was not found for the transaction: {}.", id, txn);
             txIDs.add(txn.getTxID());
           }
+          if (!iter.hasNext()) {
+            try (TableIterator<Long,
+                ? extends Table.KeyValue<Long, DeletedBlocksTransaction>> tmpIter =
+                deletedBlockLogStateManager.getReadOnlyIterator()) {
+              if (tmpIter.hasNext()) {
+                keyValue = tmpIter.next();
+                if (keyValue.getKey() != firstProcessedTransactionKey) {
+                  iter.seek(keyValue.getKey());
+                } else {
+                  break;
+                }
+              }
+            }
+          }
+        }
+        if (iter.hasNext()) {
+          lastProcessedTransactionId = iter.next().getKey();
+        } else {
+          lastProcessedTransactionId = -1L;
         }
         if (!txIDs.isEmpty()) {
           deletedBlockLogStateManager.removeTransactionsFromDB(txIDs);

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/block/TestDeletedBlockLog.java

@@ -441,6 +441,75 @@ public void testResetCount() throws Exception {
     assertEquals(30 * THREE, blocks.size());
   }
 
+
+  @Test
+  public void testSCMDelIteratorProgress() throws Exception {
+    int maxRetry = conf.getInt(OZONE_SCM_BLOCK_DELETION_MAX_RETRY, 20);
+
+    // CASE1: When all transactions are valid and available
+    // Create 8 TXs in the log.
+    int noOfTransactions = 8;
+    addTransactions(generateData(noOfTransactions), true);
+    mockContainerHealthResult(true);
+    List<DeletedBlocksTransaction> blocks;
+
+    List<Long> txIDs = new ArrayList<>();
+    int i = 1;
+    while (i < noOfTransactions) {
+      // In each iteration read two transaction, API returns all the transactions in order.
+      // 1st iteration: {1, 2}
+      // 2nd iteration: {3, 4}
+      // 3rd iteration: {5, 6}
+      // 4th iteration: {7, 8}
+      blocks = getTransactions(2 * BLOCKS_PER_TXN * THREE);
+      assertEquals(blocks.get(0).getTxID(), i++);
+      assertEquals(blocks.get(1).getTxID(), i++);
+    }
+
+    // CASE2: When some transactions are not available for delete in the current iteration,
+    // either due to max retry reach or some other issue.
+    // New transactions Id is { 9, 10, 11, 12, 13, 14, 15, 16}
+    addTransactions(generateData(noOfTransactions), true);
+    mockContainerHealthResult(true);
+
+    // Mark transaction Id 11 as reached max retry count so that it will be ignored
+    // by scm deleting service while fetching transaction for delete
+    int ignoreTransactionId = 11;
+    txIDs.add((long) ignoreTransactionId);
+    for (i = 0; i < maxRetry; i++) {
+      incrementCount(txIDs);
+    }
+    incrementCount(txIDs);
+
+    i = 9;
+    while (true) {
+      // In each iteration read two transaction.
+      // If any transaction which is not available for delete in the current iteration,
+      // it will be ignored and will be re-checked again only after complete table is read.
+      // 1st iteration: {9, 10}
+      // 2nd iteration: {12, 13} Transaction 11 is ignored here
+      // 3rd iteration: {14, 15} Transaction 11 is available here,
+      // but it will be read only when all db records are read till the end.
+      // 4th iteration: {16, 11} Since iterator reached at the end of table after reading transaction 16,
+      // Iterator starts from beginning again, and it returns transaction 11 as well
+      blocks = getTransactions(2 * BLOCKS_PER_TXN * THREE);
+      if (i == ignoreTransactionId) {
+        i++;
+      }
+      assertEquals(blocks.get(0).getTxID(), i++);
+      if (i == 17) {
+        assertEquals(blocks.get(1).getTxID(), ignoreTransactionId);
+        break;
+      }
+      assertEquals(blocks.get(1).getTxID(), i++);
+
+      if (i == 14) {
+        // Reset transaction 11 so that it will be available in scm key deleting service in the subsequent iterations.
+        resetCount(txIDs);
+      }
+    }
+  }
+
   @Test
   public void testCommitTransactions() throws Exception {
     deletedBlockLog.setScmCommandTimeoutMs(Long.MAX_VALUE);