HBASE-30335 Tolerate stale recovered.edits below durable seqid in split/merge

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManagerUtil.java

@@ -24,10 +24,14 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.ListIterator;
+import java.util.NavigableSet;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseIOException;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.ServerName;
@@ -37,9 +41,13 @@
 import org.apache.hadoop.hbase.favored.FavoredNodesManager;
 import org.apache.hadoop.hbase.master.RegionState;
 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hbase.util.FutureUtils;
 import org.apache.hadoop.hbase.wal.WALSplitUtil;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoRequest;
@@ -50,6 +58,7 @@
  */
 @InterfaceAudience.Private
 final class AssignmentManagerUtil {
+  private static final Logger LOG = LoggerFactory.getLogger(AssignmentManagerUtil.class);
   private static final int DEFAULT_REGION_REPLICA = 1;
 
   private AssignmentManagerUtil() {
@@ -294,10 +303,79 @@ static void removeNonDefaultReplicas(MasterProcedureEnv env, Stream<RegionInfo>
   }
 
   static void checkClosedRegion(MasterProcedureEnv env, RegionInfo regionInfo) throws IOException {
-    if (WALSplitUtil.hasRecoveredEdits(env.getMasterConfiguration(), regionInfo)) {
-      throw new IOException("Recovered.edits are found in Region: " + regionInfo
-        + ", abort split/merge to prevent data loss");
+    if (!WALSplitUtil.hasRecoveredEdits(env.getMasterConfiguration(), regionInfo)) {
+      return;
     }
+    // A recovered.edits file whose max seqid is <= the region's last flushed seqid is stale:
+    // its edits are already durable in HFiles. This happens e.g. when a graceful region move
+    // is followed by a WAL split of the source RS - the split creates recovered.edits for a
+    // region that has already been reopened elsewhere and flushed. Cleaning up such files here
+    // (rather than aborting split/merge) matches the tolerance HRegion itself applies at open.
+    if (tryDropStaleRecoveredEdits(env, regionInfo)) {
+      return;
+    }
+    throw new IOException("Recovered.edits are found in Region: " + regionInfo
+      + ", abort split/merge to prevent data loss");
+  }
+
+  /**
+   * Try to remove recovered.edits files that are provably below the region's last flushed seqid.
+   * @return true if, after cleanup, no recovered.edits remain for the region
+   */
+  private static boolean tryDropStaleRecoveredEdits(MasterProcedureEnv env, RegionInfo regionInfo) {
+    long durableSeqId = env.getMasterServices().getServerManager()
+      .getLastFlushedSequenceId(regionInfo.getEncodedNameAsBytes()).getLastFlushedSequenceId();
+    if (durableSeqId <= 0L) {
+      // No authoritative durability info at the master; play safe and let the caller abort.
+      return false;
+    }
+    try {
+      Configuration conf = env.getMasterConfiguration();
+      Path regionWALDir =
+        CommonFSUtils.getWALRegionDir(conf, regionInfo.getTable(), regionInfo.getEncodedName());
+      Path regionDir = FSUtils.getRegionDirFromRootDir(CommonFSUtils.getRootDir(conf), regionInfo);
+      Path wrongRegionWALDir = CommonFSUtils.getWrongWALRegionDir(conf, regionInfo.getTable(),
+        regionInfo.getEncodedName());
+      FileSystem walFs = CommonFSUtils.getWALFileSystem(conf);
+      FileSystem rootFs = CommonFSUtils.getRootDirFileSystem(conf);
+      return dropStaleEditsUnder(walFs, regionWALDir, durableSeqId, regionInfo)
+        && dropStaleEditsUnder(rootFs, regionDir, durableSeqId, regionInfo)
+        && dropStaleEditsUnder(walFs, wrongRegionWALDir, durableSeqId, regionInfo);
+    } catch (IOException e) {
+      LOG.warn("Failed to inspect recovered.edits for {}; falling back to abort", regionInfo, e);
+      return false;
+    }
+  }
+
+  private static boolean dropStaleEditsUnder(FileSystem fs, Path regionDir, long durableSeqId,
+    RegionInfo regionInfo) throws IOException {
+    NavigableSet<Path> files = WALSplitUtil.getSplitEditFilesSorted(fs, regionDir);
+    if (files.isEmpty()) {
+      return true;
+    }
+    for (Path p : files) {
+      long fileMaxSeqId;
+      try {
+        fileMaxSeqId = Long.parseLong(p.getName());
+      } catch (NumberFormatException e) {
+        LOG.warn("Non-numeric recovered.edits filename {} for {}; not dropping", p, regionInfo);
+        return false;
+      }
+      if (fileMaxSeqId > durableSeqId) {
+        LOG.info("Recovered.edits {} for {} has maxSeqId={} > durableSeqId={}; needs replay", p,
+          regionInfo, fileMaxSeqId, durableSeqId);
+        return false;
+      }
+    }
+    for (Path p : files) {
+      LOG.info("Removing stale recovered.edits {} for {} (durableSeqId={})", p, regionInfo,
+        durableSeqId);
+      if (!fs.delete(p, false)) {
+        LOG.warn("Failed to delete stale recovered.edits {} for {}", p, regionInfo);
+        return false;
+      }
+    }
+    return true;
   }
 
   /**

hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestAssignmentManagerUtil.java

@@ -18,25 +18,33 @@
 package org.apache.hadoop.hbase.master.assignment;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseIOException;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
 import org.apache.hadoop.hbase.testclassification.MasterTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.wal.WALSplitUtil;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -126,4 +134,50 @@ public void testCreateUnassignProceduresForMergeFail() throws IOException {
       .map(AM.getRegionStates()::getRegionStateNode).forEachOrdered(
         rn -> assertFalse(rn.isTransitionScheduled(), "Should have unset the proc for " + rn));
   }
+
+  @Test
+  public void testCheckClosedRegionDropsStaleRecoveredEdits() throws Exception {
+    try (Table t = UTIL.getConnection().getTable(TABLE_NAME)) {
+      for (int i = 0; i < 10; i++) {
+        t.put(new Put(Bytes.toBytes(i)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("q"),
+          Bytes.toBytes(i)));
+      }
+    }
+    UTIL.getAdmin().flush(TABLE_NAME);
+    UTIL.waitFor(30_000,
+      () -> getPrimaryRegions().stream().anyMatch(r -> ENV.getMasterServices().getServerManager()
+        .getLastFlushedSequenceId(r.getEncodedNameAsBytes()).getLastFlushedSequenceId() > 0L));
+
+    RegionInfo region = getPrimaryRegions().stream()
+      .filter(r -> ENV.getMasterServices().getServerManager()
+        .getLastFlushedSequenceId(r.getEncodedNameAsBytes()).getLastFlushedSequenceId() > 0L)
+      .findFirst().orElseThrow(() -> new AssertionError("no region has a durable seqid yet"));
+    long durable = ENV.getMasterServices().getServerManager()
+      .getLastFlushedSequenceId(region.getEncodedNameAsBytes()).getLastFlushedSequenceId();
+
+    Configuration conf = ENV.getMasterConfiguration();
+    Path walRegionDir =
+      CommonFSUtils.getWALRegionDir(conf, region.getTable(), region.getEncodedName());
+    Path editsDir = WALSplitUtil.getRegionDirRecoveredEditsDir(walRegionDir);
+    FileSystem fs = CommonFSUtils.getWALFileSystem(conf);
+    fs.mkdirs(editsDir);
+
+    Path staleFile = new Path(editsDir, String.format("%019d", 1L));
+    fs.create(staleFile).close();
+    assertTrue(fs.exists(staleFile));
+
+    AssignmentManagerUtil.checkClosedRegion(ENV, region);
+    assertFalse(fs.exists(staleFile), "stale recovered.edits should have been removed");
+
+    Path freshFile = new Path(editsDir, String.format("%019d", durable + 1000L));
+    fs.create(freshFile).close();
+    try {
+      AssignmentManagerUtil.checkClosedRegion(ENV, region);
+      fail("checkClosedRegion should have thrown for a fresh recovered.edits file");
+    } catch (IOException expected) {
+      // expected
+    } finally {
+      fs.delete(freshFile, false);
+    }
+  }
 }