(hbase) branch branch-2.6 updated: HBASE-29987 Replication position corruption when WAL file switch detected in ReplicationSourceWALReader run loop (#7909)

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhangduo pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/hbase.git


The following commit(s) were added to refs/heads/branch-2.6 by this push:
     new 7cac41a3f79 HBASE-29987 Replication position corruption when WAL file switch detected in ReplicationSourceWALReader run loop (#7909)
7cac41a3f79 is described below

commit 7cac41a3f797283e836b8d1e567bd51e383bed60
Author: Siddharth Khillon <si...@gmail.com>
AuthorDate: Thu Mar 12 20:38:47 2026 -0700

    HBASE-29987 Replication position corruption when WAL file switch detected in ReplicationSourceWALReader run loop (#7909)
    
    When ReplicationSourceWALReader.run() detects a WAL file switch via the
    switched() check, it enqueues an EOF batch but does not update
    currentPosition. If the outer loop restarts (e.g., due to
    WALEntryFilterRetryableException), the new WALEntryStream is created
    with the stale position from the old file, applied to the new file.
    This causes an infinite retry loop (EOFException: Cannot seek after EOF)
    and the corrupted position may be persisted to ZK, surviving restarts.
    
    The fix resets currentPosition to entryStream.getPosition() (which
    returns 0 after dequeueCurrentLog()) before enqueuing the EOF batch.
    
    Includes a regression test that reproduces the bug by using
    nb.capacity=1 to force EOF detection at line 153 (not inside
    readWALEntries), combined with a WALEntryFilterRetryableException on
    the first entry of the new file to trigger the outer loop restart.
    
    Co-authored-by: skhillon <sk...@hubspot.com>
    Signed-off-by: Duo Zhang <zh...@apache.org>
    (cherry picked from commit e4f9c65e80dd7d5ffb103f782f054e0cc18f5878)
---
 .../regionserver/ReplicationSourceWALReader.java   |  1 +
 .../regionserver/TestBasicWALEntryStream.java      | 52 ++++++++++++++++++++++
 2 files changed, 53 insertions(+)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
index e617fe6d016..89635ad000c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
@@ -158,6 +158,7 @@ class ReplicationSourceWALReader extends Thread {
           // first, check if we have switched a file, if so, we need to manually add an EOF entry
           // batch to the queue
           if (currentPath != null && switched(entryStream, currentPath)) {
+            currentPosition = entryStream.getPosition();
             entryBatchQueue.put(WALEntryBatch.endOfFile(currentPath));
             continue;
           }
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java
index b1e2f2c1634..b3265da3887 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestBasicWALEntryStream.java
@@ -881,6 +881,58 @@ public abstract class TestBasicWALEntryStream extends WALEntryStreamTestBase {
     }
   }
 
+  /**
+   * Verify that when a WAL file switch is detected via the switched() check in
+   * ReplicationSourceWALReader.run(), currentPosition is reset so that a subsequent
+   * WALEntryFilterRetryableException does not cause the new file to be opened at the old file's
+   * position.
+   */
+  @Test
+  public void testPositionResetOnFileSwitchWithRetryableFilter() throws Exception {
+    appendEntriesToLogAndSync(3);
+    log.rollWriter();
+    AbstractFSWAL<?> abstractWAL = (AbstractFSWAL<?>) log;
+    Waiter.waitFor(CONF, 5000,
+      (Waiter.Predicate<Exception>) () -> abstractWAL.getInflightWALCloseCount() == 0);
+    appendEntriesToLogAndSync(3);
+
+    // Batch capacity of 1 ensures EOF on WAL A is detected by hasNext() in the run loop
+    // (not inside readWALEntries), which triggers the switched() path.
+    Configuration conf = new Configuration(CONF);
+    conf.setInt("replication.source.nb.capacity", 1);
+
+    AtomicInteger totalFilterCalls = new AtomicInteger(0);
+    AtomicBoolean threwOnce = new AtomicBoolean(false);
+    WALEntryFilter filter = entry -> {
+      int callNum = totalFilterCalls.incrementAndGet();
+      if (callNum > 3 && !threwOnce.get()) {
+        threwOnce.set(true);
+        throw new WALEntryFilterRetryableException("simulated filter failure after file switch");
+      }
+      return entry;
+    };
+
+    ReplicationSource source = mockReplicationSource(false, conf);
+    when(source.isPeerEnabled()).thenReturn(true);
+    ReplicationSourceWALReader reader =
+      new ReplicationSourceWALReader(fs, conf, logQueue, 0, filter, source, fakeWalGroupId);
+    reader.start();
+
+    int totalEntries = 0;
+    long deadline = System.currentTimeMillis() + 30000;
+    while (totalEntries < 6) {
+      long remaining = deadline - System.currentTimeMillis();
+      assertTrue("Reader appears stuck - likely position corruption. Only got " + totalEntries
+        + " of 6 entries", remaining > 0);
+      WALEntryBatch batch = reader.poll(1);
+      if (batch != null && batch != WALEntryBatch.NO_MORE_DATA) {
+        totalEntries += batch.getNbEntries();
+      }
+    }
+    assertEquals(6, totalEntries);
+    assertTrue("Filter should have thrown at least once", threwOnce.get());
+  }
+
   private static class PartialWALEntryFailingWALEntryFilter implements WALEntryFilter {
     private int filteredWALEntryCount = -1;
     private int walEntryCount = 0;