HBASE-30433 Seed the master's flushedSequenceIdByRegion watermark with the region's final flushed seqid on region CLOSE

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

@@ -290,7 +290,7 @@ private void updateLastFlushedSequenceIds(ServerName sn, ServerMetrics hsl) {
     for (Entry<byte[], RegionMetrics> entry : hsl.getRegionMetrics().entrySet()) {
       byte[] encodedRegionName = Bytes.toBytes(RegionInfo.encodeRegionName(entry.getKey()));
       final long completedSeqId = entry.getValue().getCompletedSequenceId();
-      // Atomic read-modify-write so a concurrent reportRegionOpen seed (which uses
+      // Atomic read-modify-write so a concurrent advanceRegionFlushedSeq 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) -> {
@@ -1108,21 +1108,22 @@ public void removeRegion(final RegionInfo regionInfo) {
   }
 
   /**
-   * 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.
+   * Advances {@link #flushedSequenceIdByRegion} for the region to at least {@code flushedSeqId}.
+   * Called on region OPEN with the region's {@code openSeqNum} (HBASE-30335) and on region CLOSE
+   * with the region's final durable flushed seqid (HBASE-30433). Without this, the entry stays
+   * absent or stale until the hosting server's next heartbeat, so {@link #getLastFlushedSequenceId}
+   * may return {@link HConstants#NO_SEQNUM} or a too-low value and WALSplitter conservatively
+   * treats already-durable edits as unflushed - producing orphaned recovered.edits when the source
+   * server crashes soon after a drain-move or close. Uses {@code merge} with {@link Math#max} so a
+   * heartbeat-supplied value is never regressed - and, unlike {@code putIfAbsent}, a stale-low
+   * prior value is lifted to {@code flushedSeqId}. Safe because both callers pass a seqid at or
+   * below what the region has actually made durable.
    */
-  public void reportRegionOpen(final RegionInfo regionInfo, final long openSeqNum) {
-    if (openSeqNum < 0) { // NO_SEQNUM == -1
+  public void advanceRegionFlushedSeq(final RegionInfo regionInfo, final long flushedSeqId) {
+    if (flushedSeqId < 0) { // NO_SEQNUM == -1
       return;
     }
-    flushedSequenceIdByRegion.merge(regionInfo.getEncodedNameAsBytes(), openSeqNum, Math::max);
+    flushedSequenceIdByRegion.merge(regionInfo.getEncodedNameAsBytes(), flushedSeqId, Math::max);
   }
 
   public boolean isRegionInServerManagerStates(final RegionInfo hri) {

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

@@ -1230,8 +1230,17 @@ private void reportRegionStateTransition(ReportRegionStateTransitionResponse.Bui
           final RegionInfo hri = ProtobufUtil.toRegionInfo(transition.getRegionInfo(0));
           long procId =
             transition.getProcIdCount() > 0 ? transition.getProcId(0) : Procedure.NO_PROC_ID;
-          updateRegionTransition(serverNode, transition.getTransitionCode(), hri,
-            transition.hasOpenSeqNum() ? transition.getOpenSeqNum() : HConstants.NO_SEQNUM, procId);
+          long seqId =
+            transition.hasOpenSeqNum() ? transition.getOpenSeqNum() : HConstants.NO_SEQNUM;
+          // On CLOSE, seed the master's flushed-seqid watermark with the region's final flushed
+          // seqid (reported by the RS). This complements the OPEN-time seed (HBASE-30335) and
+          // narrows the graceful-close window where a WAL split of a crashed source RS could
+          // write orphaned recovered.edits for already-durable edits. Uses the same monotonic
+          // merge(Math::max) seed, so it never regresses a higher value.
+          if (transition.getTransitionCode() == TransitionCode.CLOSED && seqId >= 0) {
+            master.getServerManager().advanceRegionFlushedSeq(hri, seqId);
+          }
+          updateRegionTransition(serverNode, transition.getTransitionCode(), hri, seqId, procId);
           break;
         case READY_TO_SPLIT:
         case SPLIT:
@@ -2295,7 +2304,7 @@ void regionOpenedWithoutPersistingToMeta(RegionStateNode regionNode)
     // 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());
+    master.getServerManager().advanceRegionFlushedSeq(regionInfo, regionNode.getOpenSeqNum());
   }
 
   // should be called under the RegionStateNode lock

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -2348,7 +2348,10 @@ private boolean skipReportingTransition(final RegionStateTransitionContext conte
     builder.setServer(ProtobufUtil.toServerName(serverName));
     RegionStateTransition.Builder transition = builder.addTransitionBuilder();
     transition.setTransitionCode(code);
-    if (code == TransitionCode.OPENED && openSeqNum >= 0) {
+    // Carry the seqid for OPENED (openSeqNum) and CLOSED (durable flushed seqid). The master uses
+    // both to seed its flushedSequenceIdByRegion watermark; see HBASE-30335 and the CLOSE-time
+    // follow-up.
+    if ((code == TransitionCode.OPENED || code == TransitionCode.CLOSED) && openSeqNum >= 0) {
       transition.setOpenSeqNum(openSeqNum);
     }
     for (RegionInfo hri : hris) {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/CloseRegionHandler.java

@@ -18,7 +18,6 @@
 package org.apache.hadoop.hbase.regionserver.handler;
 
 import java.io.IOException;
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.Server;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.client.RegionInfo;
@@ -110,8 +109,12 @@ public void process() throws IOException {
       }
 
       this.rsServices.removeRegion(region, destination);
+      // Report the region's durable flushed seqid on CLOSE so the master can seed its
+      // flushedSequenceIdByRegion watermark. This complements the OPEN-time seed (HBASE-30335)
+      // and narrows the graceful-close window where a subsequent WAL split of a crashed source
+      // RS could otherwise write orphaned recovered.edits for already-durable edits.
       rsServices.reportRegionStateTransition(new RegionStateTransitionContext(TransitionCode.CLOSED,
-        HConstants.NO_SEQNUM, Procedure.NO_PROC_ID, -1, regionInfo, -1));
+        region.getMaxFlushedSeqId(), Procedure.NO_PROC_ID, -1, regionInfo, -1));
 
       // Done! Region is closed on this RS
       LOG.debug("Closed {}", region.getRegionInfo().getRegionNameAsString());

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java

@@ -25,7 +25,6 @@
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.executor.EventHandler;
 import org.apache.hadoop.hbase.executor.EventType;
@@ -145,9 +144,14 @@ public void process() throws IOException {
     }
 
     rs.removeRegion(region, destination);
+    // Report the region's durable flushed seqid on CLOSE so the master can seed its
+    // flushedSequenceIdByRegion watermark. This complements the OPEN-time seed (HBASE-30335)
+    // and narrows the graceful-close window where a subsequent WAL split of a crashed source
+    // RS could otherwise write orphaned recovered.edits for already-durable edits.
     if (
-      !rs.reportRegionStateTransition(new RegionStateTransitionContext(TransitionCode.CLOSED,
-        HConstants.NO_SEQNUM, closeProcId, -1, region.getRegionInfo(), initiatingMasterActiveTime))
+      !rs.reportRegionStateTransition(
+        new RegionStateTransitionContext(TransitionCode.CLOSED, region.getMaxFlushedSeqId(),
+          closeProcId, -1, region.getRegionInfo(), initiatingMasterActiveTime))
     ) {
       throw new IOException("Failed to report close to master: " + regionName);
     }

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

@@ -143,4 +143,58 @@ public void testFlushedSequenceIdSeededOnRegionOpen() throws IOException, Interr
       table.close();
     }
   }
+
+  /**
+   * HBASE-30335 follow-up: on graceful region CLOSE the RS reports the region's final durable
+   * flushed seqid, and the master must lift its flushedSequenceIdByRegion watermark to that value.
+   * This complements the OPEN-time seed and covers the drain-move / disable window: if the source
+   * RS crashes after the close, a subsequent WAL split must see the already-durable edits as
+   * flushed and not resurrect them as orphaned recovered.edits. Leaves unflushed edits in the
+   * memstore so that the close itself flushes and advances the region's maxFlushedSeqId; a
+   * heartbeat sent before the close cannot carry that value, so only the CLOSE report can deliver
+   * it. Disables the table (a graceful close that keeps the region - unlike delete/split/merge,
+   * disable does not call ServerManager#removeRegion) and asserts the watermark equals the
+   * post-close flushed seqid.
+   */
+  @Test
+  public void testFlushedSequenceIdSeededOnRegionClose() throws IOException, InterruptedException {
+    TableName freshTable = TableName.valueOf(getClass().getSimpleName(), "closeseed");
+    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);
+      // Leave the edits unflushed: heartbeats only report the region's maxFlushedSeqId, which
+      // stays put until the close flushes the memstore.
+      for (int i = 0; i < 3; i++) {
+        table.put(new Put(Bytes.toBytes("k" + i)).addColumn(family, Bytes.toBytes("q"),
+          Bytes.toBytes("v" + i)));
+      }
+      long preCloseFlushedSeqId = region.getMaxFlushedSeqId();
+      byte[] encodedName = region.getRegionInfo().getEncodedNameAsBytes();
+      // Gracefully close the region (disable keeps the region entry - it is not removed like a
+      // delete/split/merge would), driving a CLOSED transition that carries the flushed seqid.
+      testUtil.getAdmin().disableTable(freshTable);
+      long closeFlushedSeqId = region.getMaxFlushedSeqId();
+      assertTrue(closeFlushedSeqId > preCloseFlushedSeqId, "test setup: close flush should advance"
+        + " maxFlushedSeqId past " + preCloseFlushedSeqId + ", got " + closeFlushedSeqId);
+      RegionStoreSequenceIds ids = testUtil.getHBaseCluster().getMaster().getServerManager()
+        .getLastFlushedSequenceId(encodedName);
+      assertEquals(closeFlushedSeqId, ids.getLastFlushedSequenceId(),
+        "flushedSequenceIdByRegion should be seeded with the region's flushed seqid on CLOSE");
+    } finally {
+      table.close();
+    }
+  }
 }

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

@@ -82,35 +82,35 @@ private long lastFlushed(RegionInfo ri) {
   }
 
   @Test
-  public void testReportRegionOpenSeedsFlushedSequenceId() {
+  public void testAdvanceRegionFlushedSeqSeedsFlushedSequenceId() {
     assertEquals(HConstants.NO_SEQNUM, lastFlushed(region));
-    sm.reportRegionOpen(region, 42L);
+    sm.advanceRegionFlushedSeq(region, 42L);
     assertEquals(42L, lastFlushed(region));
   }
 
   @Test
-  public void testReportRegionOpenDoesNotRegressExistingValue() {
-    sm.reportRegionOpen(region, 100L);
+  public void testAdvanceRegionFlushedSeqDoesNotRegressExistingValue() {
+    sm.advanceRegionFlushedSeq(region, 100L);
     // A later OPEN carrying a smaller openSeqNum (e.g. after a restart replayed less) must not
     // clobber a higher watermark already seeded here or supplied by a heartbeat.
-    sm.reportRegionOpen(region, 50L);
+    sm.advanceRegionFlushedSeq(region, 50L);
     assertEquals(100L, lastFlushed(region));
   }
 
   @Test
-  public void testReportRegionOpenIgnoresNoSeqNum() {
-    sm.reportRegionOpen(region, HConstants.NO_SEQNUM);
+  public void testAdvanceRegionFlushedSeqIgnoresNoSeqNum() {
+    sm.advanceRegionFlushedSeq(region, HConstants.NO_SEQNUM);
     assertEquals(HConstants.NO_SEQNUM, lastFlushed(region));
   }
 
   @Test
-  public void testReportRegionOpenIgnoresNegativeSeqNum() {
-    sm.reportRegionOpen(region, -5L);
+  public void testAdvanceRegionFlushedSeqIgnoresNegativeSeqNum() {
+    sm.advanceRegionFlushedSeq(region, -5L);
     assertEquals(HConstants.NO_SEQNUM, lastFlushed(region));
   }
 
   /**
-   * HBASE-30335: the OPEN-time seed ({@link ServerManager#reportRegionOpen}, which uses
+   * HBASE-30335: the OPEN-time seed ({@link ServerManager#advanceRegionFlushedSeq}, which uses
    * {@code merge(Math::max)}) and the heartbeat handler ({@link ServerManager#regionServerReport})
    * both write {@code flushedSequenceIdByRegion}. A stale in-flight heartbeat from the
    * soon-to-be-dead source RS carries a lower {@code completedSequenceId}. Whatever the
@@ -142,7 +142,7 @@ public void testConcurrentStaleHeartbeatDoesNotClobberOpenSeed() throws Exceptio
         final CyclicBarrier barrier = new CyclicBarrier(2);
         Future<?> seedTask = pool.submit(() -> {
           await(barrier);
-          sm.reportRegionOpen(ri, seedSeqId);
+          sm.advanceRegionFlushedSeq(ri, seedSeqId);
         });
         Future<?> heartbeatTask = pool.submit(() -> {
           await(barrier);