[jira] [Resolved] (HBASE-30237) HBase bulkload replication causes duplicate HFile loading and compaction storm when source RPC times out

["Junegunn Choi (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/HBASE-30237?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Junegunn Choi resolved HBASE-30237.
-----------------------------------
    Resolution: Duplicate

> HBase bulkload replication causes duplicate HFile loading and compaction storm when source RPC times out
> --------------------------------------------------------------------------------------------------------
>
>                 Key: HBASE-30237
>                 URL: https://issues.apache.org/jira/browse/HBASE-30237
>             Project: HBase
>          Issue Type: Bug
>          Components: Replication
>    Affects Versions: 2.5.15, 2.6.6
>            Reporter: Nikita Awasthi
>            Priority: Major
>
> h2. Problem
> When bulkload replication is enabled (A to B), the sink cluster (B) copies HFiles from the source cluster's (A) HDFS and then performs a bulk load. If this operation takes longer than {{replication.source.shipedits.timeout}} (default 60s), the source retries the same WAL entry. Since {{HFileReplicator}} creates a new random staging directory on every invocation, the sink processes the same bulkload event multiple times -- loading the same HFiles as distinct store files repeatedly.
> This leads to:
> * *Storefile size explosion*: the sink's store file size can grow to 3x or more of the source's size.
> * *Compaction storm*: the duplicated store files trigger massive compaction work.
> * *Replication lag*: the source accumulates WAL entries while the sink is overwhelmed.
> The root cause is that HBase bulkload replication has at-least-once semantics, but {{ReplicationSink}} has no deduplication mechanism to prevent concurrent reprocessing of the same bulkload event.
> h2. Evidence (observed in production)
> In a production bulkload replication setup, the following sequence was observed:
> * A single bulkload event on the source triggered repeated BulkloadRPC calls on the sink -- more invocations than the source fired -- confirming multiple retries of the same WAL entry.
> * Sink RS CPU spiked from baseline ~2% to over 40%.
> * Sink network inbound peaked at several hundred MB/s per node.
> * Sink cluster storefile size grew to over 3x the source within ~90 minutes.
> * Compaction queue on the sink peaked at tens of thousands, requiring hours of major compaction to recover.
> * Source replication lag peaked at ~70 minutes, with WAL queue growing to tens of files per RS.
> h2. Root Cause Analysis
> The execution path is:
> {code}
> ReplicationSourceShipper.shipEdits()
>   -> HBaseInterClusterReplicationEndpoint.replicate()   [RPC with shipEditsTimeout]
>     -> ReplicationSink.replicateEntries()
>       -> HFileReplicator.replicate()
>           copyHFilesToStagingDir()      <- copies HFiles from source HDFS (slow)
>           LoadIncrementalHFiles.load()  <- bulk loads into sink table
> {code}
> {{copyHFilesToStagingDir()}} runs inside the RPC handler thread. When copying large HFiles takes longer than {{shipEditsTimeout}}, the source retries with the same WAL entry. {{HFileReplicator}} creates a new random staging directory each time, so there is no detection of an in-progress or already-completed invocation.
> Additionally, there is no rate-limiting mechanism for the HFile copy bandwidth, meaning a single large bulkload can saturate the sink cluster's network and cause cascading degradation.
> h2. Fix
> This patch addresses both issues:
> *1. Deduplication of concurrent bulkload replication*
> {{ReplicationSink}} tracks in-progress bulkload events in a {{ConcurrentHashSet}}. The key is:
> {code}
> replicationClusterId + "#" + encodedRegionName + "#" + bulkloadSeqNum
> {code}
> If a retry arrives while the same event is already being processed, it is skipped with a WARN log. The key is removed in a {{finally}} block so that subsequent retries (after the previous attempt completes or fails) are processed normally.
> Note: This prevents duplicate processing under concurrent retry scenarios. It does not change the at-least-once delivery guarantee of WAL replication.
> *2. Configurable bandwidth throttling for HFile copy*
> Introduces {{hbase.replication.bulkload.copy.bandwidth.mb}} (double, MB/s, default 0 = unlimited). A Guava {{RateLimiter}} is shared across all copy threads within a single {{HFileReplicator}} invocation, so the configured bandwidth is a per-RS ceiling regardless of {{hbase.replication.bulkload.copy.maxthreads}}.
> {{ReplicationSink}} implements {{ConfigurationObserver}} so the rate limit can be updated dynamically via configuration reload without restarting RegionServers.
> h2. Steps to Reproduce
> # Set up bulkload replication: cluster A replicates to cluster B.
> # Set {{hbase.replication.bulkload.copy.maxthreads=1}} on cluster B (or ensure HFile copy takes longer than {{replication.source.shipedits.timeout}}).
> # Perform a large bulkload on cluster A.
> # Observe that cluster B processes the same bulkload WAL entry multiple times, resulting in duplicated store files.
> h2. Expected Behavior
> Each unique bulkload WAL entry (identified by {{replicationClusterId}}, {{encodedRegionName}}, {{bulkloadSeqNum}}) should be processed at most once concurrently at the sink, even when the source retries due to RPC timeout.
> h2. Actual Behavior
> The same bulkload WAL entry is processed multiple times, creating duplicate store files and triggering compaction storm.
> h2. Additional Notes
> * Increasing {{replication.source.shipedits.timeout}} reduces the probability of retry but does not eliminate it and introduces WAL retention overhead.
> * The deduplication fix and the bandwidth throttling fix are complementary: throttling prevents network saturation that causes the timeout in the first place, while deduplication prevents the cascading failure when timeout does occur.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)