You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by vj...@apache.org on 2026/10/17 05:52:00 UTC
(hbase) branch branch-2 updated: HBASE-30433 Seed the master's flushedSequenceIdByRegion watermark with the region's final flushed seqid on region CLOSE (#8760) (#8707)
This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2 by this push:
new 6bbd1bc1c89 HBASE-30433 Seed the master's flushedSequenceIdByRegion watermark with the region's final flushed seqid on region CLOSE (#8760) (#8707)
6bbd1bc1c89 is described below
commit 6bbd1bc1c89c8b9c46ce2a1076287dbe0df12647
Author: Nirdosh Yadav <ni...@gmail.com>
AuthorDate: Sat Oct 17 11:21:53 2026 +0530
HBASE-30433 Seed the master's flushedSequenceIdByRegion watermark with the region's final flushed seqid on region CLOSE (#8760) (#8707)
Signed-off-by: Viraj Jasani <vj...@apache.org>
---
.../apache/hadoop/hbase/master/ServerManager.java | 27 +++++------
.../hbase/master/assignment/AssignmentManager.java | 15 ++++--
.../hadoop/hbase/regionserver/HRegionServer.java | 5 +-
.../regionserver/handler/CloseRegionHandler.java | 7 ++-
.../handler/UnassignRegionHandler.java | 10 ++--
.../hbase/master/TestGetLastFlushedSequenceId.java | 54 ++++++++++++++++++++++
.../hadoop/hbase/master/TestServerManager.java | 22 ++++-----
7 files changed, 107 insertions(+), 33 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
index c0debe25995..eac9caae336 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
@@ -252,7 +252,7 @@ public class ServerManager implements ConfigurationObserver {
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) -> {
@@ -1077,21 +1077,22 @@ public class ServerManager implements ConfigurationObserver {
}
/**
- * 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) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java
index fb7596930c7..d325dbde407 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java
@@ -1209,8 +1209,17 @@ public class AssignmentManager {
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:
@@ -2247,7 +2256,7 @@ public class AssignmentManager {
// 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
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index 4ebbe805c1b..3485dd6b637 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -2705,7 +2705,10 @@ public class HRegionServer extends Thread
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) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/CloseRegionHandler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/CloseRegionHandler.java
index f18e7d9ba63..c1ddd6ba5d8 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/CloseRegionHandler.java
+++ b/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 class CloseRegionHandler extends EventHandler {
}
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());
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
index 5affcd3847d..b856f5b934c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
@@ -25,7 +25,6 @@ import static org.apache.hadoop.hbase.io.hfile.CacheConfig.EVICT_BLOCKS_ON_SPLIT
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.client.RegionReplicaUtil;
import org.apache.hadoop.hbase.executor.EventHandler;
@@ -158,9 +157,14 @@ public class UnassignRegionHandler extends EventHandler {
.removeCatalogReplicationSource(region.getRegionInfo());
}
}
+ // 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);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestGetLastFlushedSequenceId.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestGetLastFlushedSequenceId.java
index 2bd2bec96e3..e0b0b8f9193 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestGetLastFlushedSequenceId.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestGetLastFlushedSequenceId.java
@@ -143,4 +143,58 @@ public class TestGetLastFlushedSequenceId {
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 {
+ MiniHBaseCluster 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();
+ }
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestServerManager.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestServerManager.java
index 9917dea9c67..d60f7e283a5 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestServerManager.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestServerManager.java
@@ -82,35 +82,35 @@ public class TestServerManager {
}
@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 class TestServerManager {
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);