HBASE-30335 Seed master flushedSequenceIdByRegion with openSeqNum on region OPEN

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

@@ -289,35 +289,50 @@ ServerName regionServerStartup(RegionServerStartupRequest request, int versionNu
   private void updateLastFlushedSequenceIds(ServerName sn, ServerMetrics hsl) {
     for (Entry<byte[], RegionMetrics> entry : hsl.getRegionMetrics().entrySet()) {
       byte[] encodedRegionName = Bytes.toBytes(RegionInfo.encodeRegionName(entry.getKey()));
-      Long existingValue = flushedSequenceIdByRegion.get(encodedRegionName);
-      long l = entry.getValue().getCompletedSequenceId();
-      // Don't let smaller sequence ids override greater sequence ids.
-      if (LOG.isTraceEnabled()) {
-        LOG.trace(Bytes.toString(encodedRegionName) + ", existingValue=" + existingValue
-          + ", completeSequenceId=" + l);
-      }
-      if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue)) {
-        flushedSequenceIdByRegion.put(encodedRegionName, l);
-      } else if (l != HConstants.NO_SEQNUM && l < existingValue) {
-        LOG.warn("RegionServer " + sn + " indicates a last flushed sequence id (" + l
-          + ") that is less than the previous last flushed sequence id (" + existingValue
-          + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
-      }
+      final long completedSeqId = entry.getValue().getCompletedSequenceId();
+      // Atomic read-modify-write so a concurrent reportRegionOpen seed (which uses
+      // merge(Math::max)) cannot be clobbered by a stale in-flight heartbeat carrying a
+      // lower completedSequenceId. Don't let smaller sequence ids override greater ones.
+      flushedSequenceIdByRegion.compute(encodedRegionName, (k, existingValue) -> {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace(Bytes.toString(k) + ", existingValue=" + existingValue + ", completeSequenceId="
+            + completedSeqId);
+        }
+        if (existingValue == null) {
+          return completedSeqId;
+        }
+        if (completedSeqId != HConstants.NO_SEQNUM && completedSeqId > existingValue) {
+          return completedSeqId;
+        }
+        if (completedSeqId != HConstants.NO_SEQNUM && completedSeqId < existingValue) {
+          LOG.warn("RegionServer " + sn + " indicates a last flushed sequence id (" + completedSeqId
+            + ") that is less than the previous last flushed sequence id (" + existingValue
+            + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
+        }
+        return existingValue;
+      });
       ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
         computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,
           () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));
       for (Entry<byte[], Long> storeSeqId : entry.getValue().getStoreSequenceId().entrySet()) {
         byte[] family = storeSeqId.getKey();
-        existingValue = storeFlushedSequenceId.get(family);
-        l = storeSeqId.getValue();
-        if (LOG.isTraceEnabled()) {
-          LOG.trace(Bytes.toString(encodedRegionName) + ", family=" + Bytes.toString(family)
-            + ", existingValue=" + existingValue + ", completeSequenceId=" + l);
-        }
-        // Don't let smaller sequence ids override greater sequence ids.
-        if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue.longValue())) {
-          storeFlushedSequenceId.put(family, l);
-        }
+        final long storeCompletedSeqId = storeSeqId.getValue();
+        storeFlushedSequenceId.compute(family, (k, existingValue) -> {
+          if (LOG.isTraceEnabled()) {
+            LOG.trace(Bytes.toString(encodedRegionName) + ", family=" + Bytes.toString(k)
+              + ", existingValue=" + existingValue + ", completeSequenceId=" + storeCompletedSeqId);
+          }
+          if (existingValue == null) {
+            return storeCompletedSeqId;
+          }
+          if (
+            storeCompletedSeqId != HConstants.NO_SEQNUM
+              && storeCompletedSeqId > existingValue.longValue()
+          ) {
+            return storeCompletedSeqId;
+          }
+          return existingValue;
+        });
       }
     }
   }
@@ -1092,6 +1107,24 @@ public void removeRegion(final RegionInfo regionInfo) {
     flushedSequenceIdByRegion.remove(encodedName);
   }
 
+  /**
+   * Called on region OPEN to seed {@link #flushedSequenceIdByRegion} with the region's
+   * {@code openSeqNum}. Without this, the entry stays absent until the hosting server's next
+   * heartbeat, so {@link #getLastFlushedSequenceId} returns {@link HConstants#NO_SEQNUM} and
+   * WALSplitter conservatively treats already-durable edits as unflushed - producing orphaned
+   * recovered.edits when the source server crashes soon after a drain-move. Uses {@code merge} with
+   * {@link Math#max} so a heartbeat-supplied value (which may reflect flushes after open) is never
+   * regressed - and, unlike {@code putIfAbsent}, a stale-low prior value is lifted to
+   * {@code openSeqNum}. Safe because at OPEN a region cannot have flushed past its own
+   * {@code openSeqNum}. See HBASE-30335.
+   */
+  public void reportRegionOpen(final RegionInfo regionInfo, final long openSeqNum) {
+    if (openSeqNum < 0) { // NO_SEQNUM == -1
+      return;
+    }
+    flushedSequenceIdByRegion.merge(regionInfo.getEncodedNameAsBytes(), openSeqNum, Math::max);
+  }
+
   public boolean isRegionInServerManagerStates(final RegionInfo hri) {
     final byte[] encodedName = hri.getEncodedNameAsBytes();
     return (storeFlushedSequenceIdsByRegion.containsKey(encodedName)

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

@@ -2287,6 +2287,10 @@ void regionOpenedWithoutPersistingToMeta(RegionStateNode regionNode)
     RegionInfo regionInfo = regionNode.getRegionInfo();
     regionStates.addRegionToServer(regionNode);
     regionStates.removeFromFailedOpen(regionInfo);
+    // HBASE-30335: seed the master's flushed sequence cache with openSeqNum so a subsequent
+    // WAL split (e.g. source RS crashes after drain-move) recognizes already-durable edits
+    // instead of writing orphaned recovered.edits.
+    master.getServerManager().reportRegionOpen(regionInfo, regionNode.getOpenSeqNum());
   }
 
   // should be called under the RegionStateNode lock

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

@@ -55,6 +55,7 @@
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
 import org.apache.hadoop.hbase.master.assignment.RegionStates;
+import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.HRegionServer;
 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
 import org.apache.hadoop.hbase.regionserver.Region;
@@ -415,6 +416,18 @@ public void makeWAL(HRegionServer hrs, List<RegionInfo> regions, int numEdits, i
     // sync every ~30k to line up with desired wal rolls
     final int syncEvery = 30 * 1024 / editSize;
     MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
+    // HBASE-30335: match the per-region seqid invariant a real WAL preserves so the splitter's
+    // openSeqNum-seeded filter doesn't drop our injected edits as already-flushed.
+    long maxOpen = 0L;
+    for (RegionInfo info : hris) {
+      HRegion r = hrs.getRegion(info.getEncodedName());
+      if (r != null) {
+        maxOpen = Math.max(maxOpen, r.getOpenSeqNum());
+      }
+    }
+    if (maxOpen > 0L) {
+      mvcc.advanceTo(maxOpen);
+    }
     if (n > 0) {
       for (int i = 0; i < numEdits; i += 1) {
         WALEdit e = new WALEdit();

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.master;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -30,6 +31,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.HRegionServer;
 import org.apache.hadoop.hbase.regionserver.Region;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
@@ -89,10 +91,12 @@ public void test() throws IOException, InterruptedException {
     Thread.sleep(2000);
     RegionStoreSequenceIds ids = testUtil.getHBaseCluster().getMaster().getServerManager()
       .getLastFlushedSequenceId(region.getRegionInfo().getEncodedNameAsBytes());
-    assertEquals(HConstants.NO_SEQNUM, ids.getLastFlushedSequenceId());
     // This will be the sequenceid just before that of the earliest edit in memstore.
     long storeSequenceId = ids.getStoreSequenceId(0).getSequenceId();
     assertTrue(storeSequenceId > 0);
+    // HBASE-30335: openSeqNum is now seeded on region OPEN, so lastFlushedSequenceId is no
+    // longer NO_SEQNUM before the first flush.
+    assertNotEquals(HConstants.NO_SEQNUM, ids.getLastFlushedSequenceId());
     testUtil.getAdmin().flush(tableName);
     Thread.sleep(2000);
     ids = testUtil.getHBaseCluster().getMaster().getServerManager()
@@ -102,4 +106,41 @@ public void test() throws IOException, InterruptedException {
     assertEquals(ids.getLastFlushedSequenceId(), ids.getStoreSequenceId(0).getSequenceId());
     table.close();
   }
+
+  /**
+   * HBASE-30335: after a region is opened - and before any user write or flush - the master's
+   * flushedSequenceIdByRegion must already contain the region's openSeqNum. Otherwise a subsequent
+   * WAL split (e.g. the hosting RS crashes before its first flush heartbeat) would treat
+   * already-durable edits as unflushed and produce orphaned recovered.edits.
+   */
+  @Test
+  public void testFlushedSequenceIdSeededOnRegionOpen() throws IOException, InterruptedException {
+    TableName freshTable = TableName.valueOf(getClass().getSimpleName(), "openseed");
+    testUtil.getAdmin()
+      .createNamespace(NamespaceDescriptor.create(freshTable.getNamespaceAsString()).build());
+    Table table = testUtil.createTable(freshTable, families);
+    try {
+      SingleProcessHBaseCluster cluster = testUtil.getMiniHBaseCluster();
+      HRegion region = null;
+      for (JVMClusterUtil.RegionServerThread rst : cluster.getRegionServerThreads()) {
+        for (HRegion r : rst.getRegionServer().getRegions(freshTable)) {
+          region = r;
+          break;
+        }
+        if (region != null) {
+          break;
+        }
+      }
+      assertNotNull(region);
+      long openSeqNum = region.getOpenSeqNum();
+      RegionStoreSequenceIds ids = testUtil.getHBaseCluster().getMaster().getServerManager()
+        .getLastFlushedSequenceId(region.getRegionInfo().getEncodedNameAsBytes());
+      assertNotEquals(HConstants.NO_SEQNUM, ids.getLastFlushedSequenceId(),
+        "flushedSequenceIdByRegion should be seeded on region OPEN (HBASE-30335)");
+      assertEquals(openSeqNum, ids.getLastFlushedSequenceId(),
+        "seeded value must equal the region's openSeqNum");
+    } finally {
+      table.close();
+    }
+  }
 }

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

@@ -20,6 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -252,10 +253,23 @@ public void testFlushedSequenceIdPersistLoad() throws Exception {
     TEST_UTIL.getMiniHBaseCluster().shutdown();
     TEST_UTIL.restartHBaseCluster(2);
     TEST_UTIL.waitUntilNoRegionsInTransition();
-    // check equality after reloading flushed sequence id map
+    // Post HBASE-30335 the master seeds flushedSequenceIdByRegion on OPEN via
+    // merge(openSeqNum, Math::max). openSeqNum is monotonic across close/open cycles, so a region
+    // reopened after cluster restart may carry a strictly higher value than what was persisted.
+    // The preserved invariant is: every region persisted at shutdown is loaded on restart, and no
+    // watermark regresses.
     Map<byte[], Long> regionMapAfter =
       TEST_UTIL.getHBaseCluster().getMaster().getServerManager().getFlushedSequenceIdByRegion();
-    assertTrue(regionMapBefore.equals(regionMapAfter));
+    assertEquals(regionMapBefore.size(), regionMapAfter.size());
+    for (Map.Entry<byte[], Long> before : regionMapBefore.entrySet()) {
+      Long after = regionMapAfter.get(before.getKey());
+      assertNotNull(after,
+        "region missing after restart: " + Bytes.toStringBinary(before.getKey()));
+      assertTrue(after >= before.getValue(),
+        "flushedSequenceId regressed across restart for region "
+          + Bytes.toStringBinary(before.getKey()) + " before=" + before.getValue() + " after="
+          + after);
+    }
   }
 
   @Test