This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new dd60203e36 PHOENIX-8004 Use BlockingWaitStrategy for the
replication-log Disruptor (#2617)
dd60203e36 is described below
commit dd60203e3634e17a0af9fa77e67ff009ca4d024a
Author: tkhurana <[email protected]>
AuthorDate: Tue Sep 15 21:06:15 2026 -0700
PHOENIX-8004 Use BlockingWaitStrategy for the replication-log Disruptor
(#2617)
The ReplicationLogGroup Disruptor was created with YieldingWaitStrategy,
which busy-spins/yields and pins a CPU core per HA group even when the ring
buffer is idle. With multiple HA groups configured this starves the region
server write path, inflating putBatch latency (client p90 for 5 writes rose
from ~30ms with no groups to 60-70ms with 7 groups).
Switch to BlockingWaitStrategy, which parks the consumer when the ring is
empty (0 CPU when idle). This matches HBase's own WAL Disruptor (FSHLog),
which uses BlockingWaitStrategy because the HDFS append/sync work dwarfs any
wakeup latency, so spinning makes no sense. Replication is off the write
critical path (max() overlap with postIndexUpdate), the ring is mostly idle,
and RS cores are contended -- the conditions where Yielding's
sub-microsecond
pickup would matter are all false.
---
.../phoenix/replication/ReplicationLogDiscovery.java | 12 ++++++------
.../apache/phoenix/replication/ReplicationLogGroup.java | 15 ++++++++++-----
2 files changed, 16 insertions(+), 11 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
index e499d16db4..0b3a8bdb2b 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java
@@ -47,12 +47,12 @@ import
org.apache.phoenix.thirdparty.com.google.common.util.concurrent.ThreadFac
* service Round-based Processing: - Files are organized into replication
rounds based on timestamps
* - Each round represents a time window (e.g., 1 minute) of replication
activity - Processing waits
* for round completion + buffer time before processing to ensure all files
are available Subclasses
- * must implement: - processFile(Path, boolean): Defines how individual
replication log files are processed -
- * createMetricsSource(): Provides metrics tracking for monitoring -
Configuration methods: Thread
- * counts, intervals, probabilities, etc. File Processing Flow: 1. Discover
new files for the
- * current round 2. Mark files as in-progress (move to in-progress directory)
3. Process each file
- * via abstract processFile() method 4. Mark successfully processed files as
completed (delete from
- * in-progress) 5. Update lastRoundProcessed to track progress
+ * must implement: - processFile(Path, boolean): Defines how individual
replication log files are
+ * processed - createMetricsSource(): Provides metrics tracking for monitoring
- Configuration
+ * methods: Thread counts, intervals, probabilities, etc. File Processing
Flow: 1. Discover new
+ * files for the current round 2. Mark files as in-progress (move to
in-progress directory) 3.
+ * Process each file via abstract processFile() method 4. Mark successfully
processed files as
+ * completed (delete from in-progress) 5. Update lastRoundProcessed to track
progress
*/
public abstract class ReplicationLogDiscovery {
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
index d77977b7a9..8f73a8c886 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
@@ -27,13 +27,13 @@ import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC_AND_FORWARD;
+import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.InsufficientCapacityException;
import com.lmax.disruptor.LifecycleAware;
import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import java.io.IOException;
@@ -795,8 +795,13 @@ public class ReplicationLogGroup {
int ringBufferSize =
conf.getInt(REPLICATION_LOG_RINGBUFFER_SIZE_KEY,
DEFAULT_REPLICATION_LOG_RINGBUFFER_SIZE);
this.disruptorExecutor = createDisruptorExecutor();
+ // BlockingWaitStrategy parks the consumer when the ring buffer is empty
(0 CPU when idle).
+ // YieldingWaitStrategy busy-spins/yields and pins a core per HA group
even with no traffic,
+ // which starves the RS write path. HBase's WAL Disruptor (FSHLog) uses
BlockingWaitStrategy
+ // for the same reason: the HDFS append/sync work dwarfs any wakeup
latency, so spinning makes
+ // no sense.
disruptor = new Disruptor<>(LogEvent.EVENT_FACTORY, ringBufferSize,
disruptorExecutor,
- ProducerType.MULTI, new YieldingWaitStrategy());
+ ProducerType.MULTI, new BlockingWaitStrategy());
eventHandler = new LogEventHandler();
eventHandler.init();
disruptor.handleEventsWith(eventHandler);
@@ -873,9 +878,9 @@ public class ReplicationLogGroup {
}
long startTime = System.nanoTime();
try {
- // ringBuffer.next() claims the next sequence number. Because we
initialize the Disruptor
- // with ProducerType.MULTI and the blocking YieldingWaitStrategy this
call WILL BLOCK if
- // the ring buffer is full, thus providing backpressure to the callers.
+ // ringBuffer.next() claims the next sequence number. Because appends
batch onto a bounded
+ // ring buffer, this call WILL BLOCK if the ring is full (producer-side
gating on the
+ // consumer's sequence), thus providing backpressure to the callers.
long sequence = ringBuffer.next();
try {
LogEvent event = ringBuffer.get(sequence);
