This is an automated email from the ASF dual-hosted git repository.
anmolnar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/master by this push:
new e309dec11f1 HBASE-29823 Control WAL flush and offset persistence from
ReplicationSourceShipper (#7617)
e309dec11f1 is described below
commit e309dec11f1f984fa2978037e2c200e11c70a3c3
Author: asolomon <[email protected]>
AuthorDate: Wed Jun 24 23:10:45 2026 +0530
HBASE-29823 Control WAL flush and offset persistence from
ReplicationSourceShipper (#7617)
* HBASE-29823 Control WAL flush and offset persistence from
ReplicationSourceShipper
Fix UT
Address review comments
Addressed review comments
Spotless fix
Move configs to Endpoint peerConfig
* Address review comments
* Fix UT issue
* Spotless fix
* Address review comments
* Minor fix
* Minor fix
* Spotless fix
* Address review comments
* Address review comments
* Spotless fix
* Address failing UT
* Add tag and category to test
* Fix compile warnings
* test fix
* Address review comment
* Move final offset persistence to normal shutdown
---------
Co-authored-by: Ankit Solomon <[email protected]>
---
.../hbase/replication/ReplicationEndpoint.java | 7 +
.../regionserver/ReplicationSource.java | 12 +
.../regionserver/ReplicationSourceShipper.java | 127 +++++++-
.../TestReplicationSourceShipperRestart.java | 327 +++++++++++++++++++++
4 files changed, 462 insertions(+), 11 deletions(-)
diff --git
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java
index 5edd5b3e8c9..3407189fcf6 100644
---
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java
+++
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java
@@ -283,4 +283,11 @@ public interface ReplicationEndpoint extends
ReplicationPeerConfigListener {
* @throws IllegalStateException if this service's state isn't FAILED.
*/
Throwable failureCause();
+
+ /**
+ * Hook invoked before persisting replication offsets. Eg: Buffered
endpoints can flush/close WALs
+ * here.
+ */
+ default void beforePersistingReplicationOffset() throws IOException {
+ }
}
diff --git
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
index 4e122ef5e8b..5545a1fa0f5 100644
---
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
+++
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
@@ -866,4 +866,16 @@ public class ReplicationSource implements
ReplicationSourceInterface {
long getSleepForRetries() {
return sleepForRetries;
}
+
+ void restartShipper(String walGroupId, ReplicationSourceShipper oldWorker) {
+ boolean removed = workerThreads.remove(walGroupId, oldWorker);
+ if (!removed) {
+ // Worker was already replaced (e.g. concurrent restart)
+ LOG.debug("Skip restart for walGroupId={} as worker already replaced",
walGroupId);
+ return;
+ }
+
+ tryStartNewShipper(walGroupId);
+ }
+
}
diff --git
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
index d05e4fed045..deaa98e71ab 100644
---
a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
+++
b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
@@ -74,6 +74,18 @@ public class ReplicationSourceShipper extends Thread {
private final int DEFAULT_TIMEOUT = 20000;
private final int getEntriesTimeout;
private final int shipEditsTimeout;
+ private long accumulatedSizeSinceLastUpdate = 0L;
+ private long lastOffsetUpdateTime = EnvironmentEdgeManager.currentTime();
+ private final long offsetUpdateIntervalMs;
+ private final long offsetUpdateSizeThresholdBytes;
+ private WALEntryBatch lastShippedBatch;
+
+ private static final String OFFSET_UPDATE_INTERVAL_MS_KEY =
+ "hbase.replication.shipper.offset.update.interval.ms";
+ private static final String OFFSET_UPDATE_SIZE_THRESHOLD_KEY =
+ "hbase.replication.shipper.offset.update.size.threshold";
+ private static final long DEFAULT_OFFSET_UPDATE_INTERVAL_MS = Long.MAX_VALUE;
+ private static final long DEFAULT_OFFSET_UPDATE_SIZE_THRESHOLD = -1L;
public ReplicationSourceShipper(Configuration conf, String walGroupId,
ReplicationSource source,
ReplicationSourceWALReader walReader) {
@@ -90,6 +102,10 @@ public class ReplicationSourceShipper extends Thread {
this.conf.getInt("replication.source.getEntries.timeout",
DEFAULT_TIMEOUT);
this.shipEditsTimeout =
this.conf.getInt(HConstants.REPLICATION_SOURCE_SHIPEDITS_TIMEOUT,
HConstants.REPLICATION_SOURCE_SHIPEDITS_TIMEOUT_DFAULT);
+ this.offsetUpdateIntervalMs =
+ conf.getLong(OFFSET_UPDATE_INTERVAL_MS_KEY,
DEFAULT_OFFSET_UPDATE_INTERVAL_MS);
+ this.offsetUpdateSizeThresholdBytes =
+ conf.getLong(OFFSET_UPDATE_SIZE_THRESHOLD_KEY,
DEFAULT_OFFSET_UPDATE_SIZE_THRESHOLD);
}
@Override
@@ -106,9 +122,23 @@ public class ReplicationSourceShipper extends Thread {
continue;
}
try {
- WALEntryBatch entryBatch = entryReader.poll(getEntriesTimeout);
+ // check time-based offset persistence
+ if (shouldPersistLogPosition()) {
+ persistLogPosition();
+ }
+
+ long pollTimeout = getEntriesTimeout;
+ if (offsetUpdateIntervalMs != Long.MAX_VALUE) {
+ long elapsed = EnvironmentEdgeManager.currentTime() -
lastOffsetUpdateTime;
+ long remaining = offsetUpdateIntervalMs - elapsed;
+ if (remaining > 0) {
+ pollTimeout = Math.min(getEntriesTimeout, remaining);
+ }
+ }
+ WALEntryBatch entryBatch = entryReader.poll(pollTimeout);
LOG.debug("Shipper from source {} got entry batch from reader: {}",
source.getQueueId(),
entryBatch);
+
if (entryBatch == null) {
continue;
}
@@ -119,13 +149,28 @@ public class ReplicationSourceShipper extends Thread {
shipEdits(entryBatch);
}
} catch (InterruptedException | ReplicationRuntimeException e) {
- // It is interrupted and needs to quit.
LOG.warn("Interrupted while waiting for next replication entry batch",
e);
Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ // During source shutdown / peer removal we can see interrupted IOEs
+ // from replication queue updates. Do not restart in this case.
+ if (!source.isSourceActive() || isInterrupted() ||
!source.isPeerEnabled()) {
+ LOG.info("Ignoring persist failure during shutdown for
walGroupId={}", walGroupId, e);
+ break;
+ }
+ LOG.error("Shipper {} failed to persist offset, restarting",
walGroupId, e);
+ abortAndRestart(e);
+ return;
}
}
+
// If the worker exits run loop without finishing its task, mark it as
stopped.
if (!isFinished()) {
+ try {
+ persistLogPosition();
+ } catch (Exception e) {
+ LOG.error("Failed persisting final offset for walGroupId={}",
walGroupId, e);
+ }
setWorkerState(WorkerState.STOPPED);
} else {
source.removeWorker(this);
@@ -133,7 +178,12 @@ public class ReplicationSourceShipper extends Thread {
}
}
- private void noMoreData() {
+ private void noMoreData() throws IOException {
+ // Flush any outstanding replication offset before marking the queue
finished.
+ // Offset persistence may be delayed by size/time thresholds, so ensure the
+ // latest replicated position is stored before transitioning to FINISHED
state.
+ persistLogPosition();
+
if (source.isRecovered()) {
LOG.debug("Finished recovering queue for group {} of peer {}",
walGroupId,
source.getQueueId());
@@ -151,16 +201,20 @@ public class ReplicationSourceShipper extends Thread {
/**
* Do the shipping logic
*/
- private void shipEdits(WALEntryBatch entryBatch) {
+ void shipEdits(WALEntryBatch entryBatch) throws IOException {
List<Entry> entries = entryBatch.getWalEntries();
int sleepMultiplier = 0;
+ int currentSize = (int) entryBatch.getHeapSize();
+ MetricsSource metrics = source.getSourceMetrics();
+ if (metrics != null && !entries.isEmpty()) {
+ metrics.setTimeStampNextToReplicate(entries.get(entries.size() -
1).getKey().getWriteTime());
+ }
if (entries.isEmpty()) {
- updateLogPosition(entryBatch);
+ // empty batch may mean WAL boundary advancement
+ lastShippedBatch = entryBatch;
+ persistLogPosition();
return;
}
- int currentSize = (int) entryBatch.getHeapSize();
- source.getSourceMetrics()
- .setTimeStampNextToReplicate(entries.get(entries.size() -
1).getKey().getWriteTime());
while (isActive()) {
try {
try {
@@ -195,8 +249,6 @@ public class ReplicationSourceShipper extends Thread {
cleanUpHFileRefs(entry.getEdit());
LOG.trace("shipped entry {}: ", entry);
}
- // Log and clean up WAL logs
- updateLogPosition(entryBatch);
// offsets totalBufferUsed by deducting shipped batchSize (excludes
bulk load size)
// this sizeExcludeBulkLoad has to use same calculation that when
calling
@@ -227,9 +279,50 @@ public class ReplicationSourceShipper extends Thread {
}
}
}
+
+ accumulatedSizeSinceLastUpdate += currentSize;
+ lastShippedBatch = entryBatch;
+ if (shouldPersistLogPosition()) {
+ persistLogPosition();
+ }
+ }
+
+ private boolean shouldPersistLogPosition() {
+ if (lastShippedBatch == null) {
+ return false;
+ }
+ // Default behaviour to update offset immediately after replicate()
+ if (offsetUpdateSizeThresholdBytes == -1 && offsetUpdateIntervalMs ==
Long.MAX_VALUE) {
+ return true;
+ }
+
+ return (accumulatedSizeSinceLastUpdate >= offsetUpdateSizeThresholdBytes)
+ || (EnvironmentEdgeManager.currentTime() - lastOffsetUpdateTime >=
offsetUpdateIntervalMs);
+ }
+
+ void persistLogPosition() throws IOException {
+ if (lastShippedBatch == null) {
+ return;
+ }
+
+ if (!source.isSourceActive() || isInterrupted() ||
!source.isPeerEnabled()) {
+ LOG.debug("Skip persistLogPosition for inactive/stopping source");
+ return;
+ }
+
+ ReplicationEndpoint endpoint = source.getReplicationEndpoint();
+ if (endpoint != null) {
+ endpoint.beforePersistingReplicationOffset();
+ }
+
+ // Log and clean up WAL logs
+ updateLogPosition(lastShippedBatch);
+ accumulatedSizeSinceLastUpdate = 0;
+ lastShippedBatch = null;
+ lastOffsetUpdateTime = EnvironmentEdgeManager.currentTime();
}
- private void cleanUpHFileRefs(WALEdit edit) throws IOException {
+ void cleanUpHFileRefs(WALEdit edit) throws IOException {
String peerId = source.getPeerId();
if (peerId.contains("-")) {
// peerClusterZnode will be in the form peerId + "-" + rsZNode.
@@ -359,4 +452,16 @@ public class ReplicationSourceShipper extends Thread {
long getSleepForRetries() {
return sleepForRetries;
}
+
+ // Restart worker so replication resumes from last persisted offset.
+ void abortAndRestart(Throwable cause) {
+ LOG.warn("Restarting shipper for walGroupId={}", walGroupId, cause);
+ if (!source.isSourceActive() || !source.isPeerEnabled() ||
isInterrupted()) {
+ LOG.warn("abortAndRestart SKIPPED walGroupId={}, thread={}", walGroupId,
+ Thread.currentThread().getName());
+ return;
+ }
+ setWorkerState(WorkerState.STOPPED);
+ source.restartShipper(walGroupId, this);
+ }
}
diff --git
a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceShipperRestart.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceShipperRestart.java
new file mode 100644
index 00000000000..bda8ea7efa2
--- /dev/null
+++
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceShipperRestart.java
@@ -0,0 +1,327 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Collections;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
+import org.apache.hadoop.hbase.replication.ReplicationQueueId;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.wal.WAL.Entry;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.hadoop.hbase.wal.WALKeyImpl;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.mockito.InOrder;
+
+@Tag(ReplicationTests.TAG)
+@Tag(SmallTests.TAG)
+public class TestReplicationSourceShipperRestart {
+
+ private ReplicationSource source;
+ private ReplicationSourceWALReader reader;
+ private MetricsSource metrics;
+ private ReplicationEndpoint endpoint;
+
+ private static final Path PATH = new Path("/test");
+
+ @BeforeEach
+ public void setup() {
+ source = mock(ReplicationSource.class);
+ reader = mock(ReplicationSourceWALReader.class);
+ metrics = mock(MetricsSource.class);
+ endpoint = mock(ReplicationEndpoint.class);
+
+ when(source.isPeerEnabled()).thenReturn(true);
+ when(source.isSourceActive()).thenReturn(true);
+ when(source.getSourceMetrics()).thenReturn(metrics);
+ when(source.getReplicationEndpoint()).thenReturn(endpoint);
+
+ ReplicationQueueId qid = mock(ReplicationQueueId.class);
+ when(source.getQueueId()).thenReturn(qid);
+
+ when(endpoint.replicate(any())).thenReturn(true);
+
+ doNothing().when(source).logPositionAndCleanOldLogs(any());
+ }
+
+ private WALEntryBatch emptyBatch() {
+ WALEntryBatch b = mock(WALEntryBatch.class);
+ when(b.getWalEntries()).thenReturn(Collections.emptyList());
+ when(b.getHeapSize()).thenReturn(100L);
+ when(b.getLastWalPosition()).thenReturn(1L);
+ when(b.getLastWalPath()).thenReturn(PATH);
+ return b;
+ }
+
+ private WALEntryBatch batch(long size) {
+ WALEntryBatch b = mock(WALEntryBatch.class);
+
+ Entry e = mock(Entry.class);
+ WALEdit edit = mock(WALEdit.class);
+ WALKeyImpl key = mock(WALKeyImpl.class);
+
+ when(key.getWriteTime()).thenReturn(1L);
+ when(key.getTableName()).thenReturn(TableName.valueOf("t"));
+
+ when(e.getKey()).thenReturn(key);
+ when(e.getEdit()).thenReturn(edit);
+
+ when(b.getWalEntries()).thenReturn(Collections.singletonList(e));
+ when(b.getHeapSize()).thenReturn(size);
+ when(b.getLastWalPosition()).thenReturn(1L);
+ when(b.getLastWalPath()).thenReturn(PATH);
+
+ return b;
+ }
+
+ /**
+ * Light shipper → skips persist logic
+ */
+ private ReplicationSourceShipper lightShipper(Configuration conf) {
+ return new ReplicationSourceShipper(conf, "group", source, reader) {
+ int loops = 0;
+
+ @Override
+ protected boolean isActive() {
+ return loops++ < 2;
+ }
+
+ @Override
+ protected void cleanUpHFileRefs(WALEdit edit) {
+ }
+
+ @Override
+ void persistLogPosition() {
+ // skip heavy logic
+ }
+ };
+ }
+
+ /**
+ * Real shipper → executes persist logic
+ */
+ private ReplicationSourceShipper realShipper(Configuration conf) {
+ return new ReplicationSourceShipper(conf, "group", source, reader) {
+ int loops = 0;
+
+ @Override
+ protected boolean isActive() {
+ return loops++ < 2;
+ }
+
+ @Override
+ protected void cleanUpHFileRefs(WALEdit edit) {
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // Restart
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testAbortAndRestart() {
+ ReplicationSourceShipper s =
+ new ReplicationSourceShipper(new Configuration(), "group", source,
reader);
+
+ s.abortAndRestart(new IOException());
+
+ verify(source).restartShipper("group", s);
+ }
+
+ // ------------------------------------------------------------------------
+ // Empty batch
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testEmptyBatchTriggersPersist() throws Exception {
+ ReplicationSourceShipper s = spy(lightShipper(new Configuration()));
+
+ s.shipEdits(emptyBatch());
+
+ verify(s).persistLogPosition();
+ }
+
+ // ------------------------------------------------------------------------
+ // Default persist
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testDefaultImmediatePersist() throws Exception {
+ ReplicationSourceShipper s = spy(lightShipper(new Configuration()));
+
+ s.shipEdits(batch(50));
+
+ verify(s).persistLogPosition();
+ }
+
+ // ------------------------------------------------------------------------
+ // Size threshold
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testSizeThreshold() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setLong("hbase.replication.shipper.offset.update.size.threshold",
100);
+
+ ReplicationSourceShipper s = spy(lightShipper(conf));
+
+ s.shipEdits(batch(40));
+ verify(s, never()).persistLogPosition();
+
+ s.shipEdits(batch(70));
+ verify(s, times(1)).persistLogPosition();
+ }
+
+ // ------------------------------------------------------------------------
+ // Time threshold
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testTimeThreshold() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setLong("hbase.replication.shipper.offset.update.interval.ms", 1);
+
+ ReplicationSourceShipper s = spy(lightShipper(conf));
+
+ s.shipEdits(batch(10));
+ Thread.sleep(20);
+ s.shipEdits(batch(10));
+
+ verify(s, atLeastOnce()).persistLogPosition();
+ }
+
+ // ------------------------------------------------------------------------
+ // Hook test (FIXED)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testBeforePersistHookCalled() throws Exception {
+ ReplicationSourceShipper s = realShipper(new Configuration());
+
+ s.shipEdits(emptyBatch());
+
+ verify(endpoint, atLeastOnce()).beforePersistingReplicationOffset();
+ }
+
+ // ------------------------------------------------------------------------
+ // Ordering test (FIXED)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testBeforePersistCalledBeforeUpdate() throws Exception {
+ ReplicationSourceShipper s = realShipper(new Configuration());
+
+ InOrder order = inOrder(endpoint, source);
+
+ s.shipEdits(batch(100));
+
+ order.verify(endpoint).beforePersistingReplicationOffset();
+ order.verify(source).logPositionAndCleanOldLogs(any());
+ }
+
+ // ------------------------------------------------------------------------
+ // Persist failure
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testPersistFailureThrows() throws Exception {
+ doThrow(new
IOException()).when(endpoint).beforePersistingReplicationOffset();
+
+ ReplicationSourceShipper shipper =
+ new ReplicationSourceShipper(new Configuration(), "group", source,
reader);
+
+ assertThrows(IOException.class, () -> shipper.shipEdits(emptyBatch()));
+ }
+
+ // ------------------------------------------------------------------------
+ // Restart via run
+ // ------------------------------------------------------------------------
+
+ @Test
+ @Timeout(30)
+ public void testPersistFailureTriggersRestart() throws Exception {
+
+ WALEntryBatch batch = emptyBatch();
+
+ when(reader.poll(anyLong())).thenReturn(batch).thenThrow(new
InterruptedException());
+
+ doThrow(new
IOException()).when(endpoint).beforePersistingReplicationOffset();
+
+ ReplicationSourceShipper shipper =
+ new ReplicationSourceShipper(new Configuration(), "group", source,
reader) {
+ int loops = 0;
+
+ @Override
+ protected boolean isActive() {
+ return loops++ < 2;
+ }
+ };
+
+ shipper.start();
+ shipper.join();
+
+ verify(source, atLeastOnce()).restartShipper(eq("group"), eq(shipper));
+ }
+
+ // ------------------------------------------------------------------------
+ // Normal run
+ // ------------------------------------------------------------------------
+
+ @Test
+ @Timeout(30)
+ public void testRunNormalFlow() throws Exception {
+
+ WALEntryBatch batch = emptyBatch();
+
+ when(reader.poll(anyLong())).thenReturn(batch).thenThrow(new
InterruptedException());
+
+ ReplicationSourceShipper s = lightShipper(new Configuration());
+
+ ReplicationSourceShipper spy = spy(s);
+ doReturn(true, true, false).when(spy).isActive();
+
+ spy.start();
+ spy.join();
+
+ verify(source, never()).restartShipper(any(), any());
+ }
+}
