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 fced737bca90..52bab6945dd8 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 @@ -290,7 +290,7 @@ private void updateLastFlushedSequenceIds(ServerName sn, ServerMetrics hsl) { for (Entry 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) { 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 0fe327fb83da..0f5d9e3a4908 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 @@ -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 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 e11b334442f1..30105cd82b17 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 @@ -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) { 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 f18e7d9ba635..c1ddd6ba5d8c 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 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()); 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 815d8bed3322..542e04d117e5 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 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); } 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 f2b76a07065e..a23cdaf6db2a 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 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(); + } + } } 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 9917dea9c67f..d60f7e283a5b 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 @@ 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);