Limit the cost in one run of block integrity check

### What changes are proposed in this pull request?

Avoid a full scan if there is already a lot of inode with integrity
problem

### Why are the changes needed?
Trying to limit the memory consumed at one scan if the inodes who have
the integrity problem.

### Does this PR introduce any user facing changes?
NA

pr-link: Alluxio#15892
change-id: cid-12ce0fe762679485777827f525f3391f1c82f8e7

core/common/src/main/java/alluxio/conf/PropertyKey.java

@@ -1933,6 +1933,14 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.IGNORE)
           .setScope(Scope.MASTER)
           .build();
+  public static final PropertyKey MASTER_BLOCK_SCAN_INVALID_BATCH_MAX_SIZE =
+      intBuilder(Name.MASTER_BLOCK_SCAN_INVALID_BATCH_MAX_SIZE)
+          .setDefaultValue(10_000_000)
+          .setDescription("The invalid block max batch size when the master is scanning the invalid"
+              + " blocks, minus number means no limit.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.IGNORE)
+          .setScope(Scope.MASTER)
+          .build();
   public static final PropertyKey MASTER_DAILY_BACKUP_ENABLED =
       booleanBuilder(Name.MASTER_DAILY_BACKUP_ENABLED)
           .setDefaultValue(false)
@@ -6909,6 +6917,8 @@ public static final class Name {
         "alluxio.master.backup.state.lock.interrupt.cycle.interval";
     public static final String MASTER_BACKUP_SUSPEND_TIMEOUT =
         "alluxio.master.backup.suspend.timeout";
+    public static final String MASTER_BLOCK_SCAN_INVALID_BATCH_MAX_SIZE =
+        "alluxio.master.block.scan.invalid.batch.max.size";
     public static final String MASTER_SHELL_BACKUP_STATE_LOCK_GRACE_MODE =
         "alluxio.master.shell.backup.state.lock.grace.mode";
     public static final String MASTER_SHELL_BACKUP_STATE_LOCK_TRY_DURATION =

core/server/master/src/main/java/alluxio/master/block/DefaultBlockMaster.java

@@ -728,9 +728,10 @@ public void removeBlocks(Collection<Long> blockIds, boolean delete) throws Unava
   @Override
   public void validateBlocks(Function<Long, Boolean> validator, boolean repair)
       throws UnavailableException {
+    long scanLimit = Configuration.getInt(PropertyKey.MASTER_BLOCK_SCAN_INVALID_BATCH_MAX_SIZE);
     List<Long> invalidBlocks = new ArrayList<>();
     try (CloseableIterator<Block> iter = mBlockMetaStore.getCloseableIterator()) {
-      while (iter.hasNext()) {
+      while (iter.hasNext() && (invalidBlocks.size() < scanLimit || scanLimit < 0)) {
         long id = iter.next().getId();
         if (!validator.apply(id)) {
           invalidBlocks.add(id);