HBASE-29823 Control WAL flush and offset persistence from ReplicationSourceShipper #7617
Conversation
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
|
@Apache9 @anmolnar @vinayakphegde Please review this PR as followup to preview PR #7591 |
|
🎊 +1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
There was a problem hiding this comment.
Pull request overview
This PR adds staged flush support to ReplicationSourceShipper to enable safe offset persistence for buffering ReplicationEndpoints (like ContinuousBackupReplicationEndpoint). The shipper now tracks staged WAL size and flush timing, only persisting offsets after buffered data is durably flushed.
Changes:
- Added buffer management hooks (
getMaxBufferSize(),maxFlushInterval(),beforePersistingReplicationOffset()) to ReplicationEndpoint interface - Modified ReplicationSourceShipper to stage WAL entries and defer offset persistence based on buffer size and time thresholds
- Added test to verify
beforePersistingReplicationOffset()is not called for empty batches
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 7 comments.
| File | Description |
|---|---|
| ReplicationEndpoint.java | Added three default methods to support buffered replication: getMaxBufferSize(), maxFlushInterval(), and beforePersistingReplicationOffset() |
| ReplicationSourceShipper.java | Modified to track staged WAL entries and defer offset persistence until buffer/timeout thresholds are met |
| TestReplicationSourceShipperBufferedFlush.java | Added test to verify beforePersistingReplicationOffset() behavior with empty batches |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| for (Entry entry : entriesForCleanUpHFileRefs) { | ||
| try { | ||
| cleanUpHFileRefs(entry.getEdit()); | ||
| } catch (IOException e) { |
There was a problem hiding this comment.
The error handling for IOException has been changed to catch and log the exception instead of propagating it. This silently suppresses IOException failures during cleanup, which could hide serious issues like file system problems. If cleanup failures should be non-fatal, this should be explicitly documented, or consider at least incrementing a failure metric to track these errors.
| } catch (IOException e) { | |
| } catch (IOException e) { | |
| // Cleanup failures are intentionally treated as non-fatal: replication has already | |
| // succeeded for these entries, so we log the failure and continue. |
| if (stagedWalSize == 0 || lastShippedBatch == null) { | ||
| return false; | ||
| } | ||
| return (stagedWalSize >= source.getReplicationEndpoint().getMaxBufferSize()) |
There was a problem hiding this comment.
The condition checks if stagedWalSize is greater than or equal to getMaxBufferSize(), but getMaxBufferSize() can return -1 for non-buffering endpoints (as documented in ReplicationEndpoint). This means the comparison 'stagedWalSize >= -1' would always be true when stagedWalSize > 0, causing immediate flushes for non-buffering endpoints. While this preserves existing behavior, the intent is unclear and could be confusing. Consider explicitly checking for -1 to make the logic more explicit.
| return (stagedWalSize >= source.getReplicationEndpoint().getMaxBufferSize()) | |
| long maxBufferSize = source.getReplicationEndpoint().getMaxBufferSize(); | |
| // For non-buffering endpoints, getMaxBufferSize() returns a negative value (e.g., -1). | |
| // In that case, we always trigger a flush based on size as soon as there is staged data. | |
| boolean sizeBasedFlush = | |
| (maxBufferSize < 0) || (stagedWalSize >= maxBufferSize); | |
| return sizeBasedFlush |
| // Allow loop to run | ||
| Waiter.waitFor(conf, 3000, () -> true); | ||
|
|
||
| shipper.interrupt(); |
There was a problem hiding this comment.
The test uses a fixed wait time of 3000ms with an empty lambda that always returns true (Waiter.waitFor(conf, 3000, () -> true)). This just sleeps unconditionally and doesn't actually verify any condition. The test should wait for a meaningful condition, such as verifying that the shipper has processed the batch or checking that the thread has reached a specific state. This makes the test timing-dependent and unreliable.
| // Allow loop to run | |
| Waiter.waitFor(conf, 3000, () -> true); | |
| shipper.interrupt(); | |
| // Wait until the shipper thread has finished processing the batch | |
| Waiter.waitFor(conf, 3000, () -> !shipper.isAlive()); |
| batch.setLastWalPosition(1L); | ||
| // no entries, no heap size | ||
|
|
||
| Mockito.when(walReader.take()).thenReturn(batch).thenReturn(null); |
There was a problem hiding this comment.
The mock is configured to stub 'walReader.take()' but the ReplicationSourceShipper actually calls 'entryReader.poll(getEntriesTimeout)' (line 118 in ReplicationSourceShipper.java). This means the mock configuration has no effect, and the test will call the unmocked poll() method which will return null, causing the shipper to skip processing any batches. The test should mock 'poll(anyInt())' instead of 'take()'.
| Mockito.when(walReader.take()).thenReturn(batch).thenReturn(null); | |
| Mockito.when(walReader.poll(Mockito.anyInt())).thenReturn(batch).thenReturn(null); |
| // ContinuousBackupReplicationEndpoint | ||
| // and -1 for other ReplicationEndpoint since they don't buffer. | ||
| // For other ReplicationEndpoint, everytime a WALEntryBatch is shipped, we update replication | ||
| // offset. Please check ReplicationSourceShipper#shouldFlushStagedWal() |
There was a problem hiding this comment.
The comment references 'shouldFlushStagedWal()' but the actual method name in ReplicationSourceShipper is 'shouldPersistLogPosition()'. This inconsistency will confuse developers trying to understand the interaction between these components.
| // ContinuousBackupReplicationEndpoint | ||
| // and Long.MAX_VALUE for other ReplicationEndpoint since they don't buffer. | ||
| // For other ReplicationEndpoint, everytime a WALEntryBatch is shipped, we update replication | ||
| // offset. Please check ReplicationSourceShipper#shouldFlushStagedWal() |
There was a problem hiding this comment.
The comment references 'shouldFlushStagedWal()' but the actual method name in ReplicationSourceShipper is 'shouldPersistLogPosition()'. This inconsistency will confuse developers trying to understand the interaction between these components.
| LOG.info("Running ReplicationSourceShipper Thread for wal group: {}", this.walGroupId); | ||
| // Loop until we close down | ||
| while (isActive()) { | ||
| // check if flush needed for WAL backup, this is need for timeout based flush |
There was a problem hiding this comment.
The comment has a typo: 'need' should be 'needed'. Should read "this is needed for timeout based flush".
| // check if flush needed for WAL backup, this is need for timeout based flush | |
| // check if flush needed for WAL backup, this is needed for timeout based flush |
taklwu
left a comment
taklwu
left a comment
There was a problem hiding this comment.
can you also check if these two failures are related?
TestHRegionWithInMemoryFlush.testParallelIncrementWithMemStoreFlush
TestTags.testFlushAndCompactionwithCombinations
| return -1; | ||
| } | ||
|
|
||
| // WAL entries are buffered in ContinuousBackupReplicationEndpoint before flushing to WAL backup |
There was a problem hiding this comment.
nit: ContinuousBackupReplicationEndpoint is part of #7591 and it's yet committing to master, should we mention early in this change?
| } catch (IOException e) { | ||
| LOG.warn("{} threw unknown exception:", | ||
| source.getReplicationEndpoint().getClass().getName(), e); | ||
| } |
There was a problem hiding this comment.
nit: will this be a behavior change because previously when cleanUpHFileRefs failed, it's throwing thru the function but here we're logging it only.
4 participants
ReplicationSourceShipper currently persists replication offsets immediately after shipping WAL batches. This is unsafe for buffering ReplicationEndpoints (e.g. ContinuousBackupReplicationEndpoint) that stage WAL entries before durably flushing them.
This change adds staged flush support by:
Introducing beforePersistingReplicationOffset() to allow endpoints to flush staged WAL data before offsets are persisted
Default implementations preserve existing behaviour for non-buffering endpoints, which continue to persist offsets immediately.
Added TestReplicationSourceShipperBufferedFlush