This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 1c7e356f00 PHOENIX-7986 : Replace IndexCDCConsumer's exponential idle
backoff with a fixed poll interval (#2597)
1c7e356f00 is described below
commit 1c7e356f0095f741059f8af027d775a85cbdbc18
Author: Palash Chauhan <[email protected]>
AuthorDate: Fri Aug 14 15:11:01 2026 -0700
PHOENIX-7986 : Replace IndexCDCConsumer's exponential idle backoff with a
fixed poll interval (#2597)
---
.../phoenix/hbase/index/IndexCDCConsumer.java | 26 ++++++++++++++++------
.../phoenix/end2end/IndexCDCConsumerLagIT.java | 1 -
2 files changed, 19 insertions(+), 8 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
index 25b0e18f26..46423f0d87 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexCDCConsumer.java
@@ -114,6 +114,13 @@ public class IndexCDCConsumer implements Runnable {
"phoenix.index.cdc.consumer.poll.interval.ms";
private static final long DEFAULT_POLL_INTERVAL_MS = 500;
+ /**
+ * The interval in milliseconds between polls when the previous poll found
no new events.
+ */
+ public static final String INDEX_CDC_CONSUMER_IDLE_POLL_INTERVAL_MS =
+ "phoenix.index.cdc.consumer.idle.poll.interval.ms";
+ private static final long DEFAULT_IDLE_POLL_INTERVAL_MS = 4000;
+
/**
* The time buffer in milliseconds subtracted from current time when
querying CDC mutations to
* help avoid reading mutations that are too recent.
@@ -157,6 +164,7 @@ public class IndexCDCConsumer implements Runnable {
private final long startupDelayMs;
private final int batchSize;
private final long pollIntervalMs;
+ private final long idlePollIntervalMs;
private final long timestampBufferMs;
private final int maxDataVisibilityRetries;
private final long parentProgressPauseMs;
@@ -237,6 +245,10 @@ public class IndexCDCConsumer implements Runnable {
this.batchSize = baseBatchSize + jitter;
this.pollIntervalMs =
config.getLong(INDEX_CDC_CONSUMER_POLL_INTERVAL_MS,
DEFAULT_POLL_INTERVAL_MS);
+ long baseIdlePollInterval =
+ config.getLong(INDEX_CDC_CONSUMER_IDLE_POLL_INTERVAL_MS,
DEFAULT_IDLE_POLL_INTERVAL_MS);
+ this.idlePollIntervalMs =
+ baseIdlePollInterval +
ThreadLocalRandom.current().nextLong(baseIdlePollInterval / 5 + 1);
this.timestampBufferMs =
config.getLong(INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS,
DEFAULT_TIMESTAMP_BUFFER_MS);
this.maxDataVisibilityRetries =
config.getInt(INDEX_CDC_CONSUMER_MAX_DATA_VISIBILITY_RETRIES,
@@ -449,11 +461,11 @@ public class IndexCDCConsumer implements Runnable {
lagEmissionEnabled = true;
LOG.info(
"IndexCDCConsumer started for table {} region {}"
- + " [batchSize: {}, pollIntervalMs: {}, timestampBufferMs: {},
startupDelayMs: {},"
- + " pause: {}, maxDataVisibilityRetries: {}, parentProgressPauseMs:
{},"
- + " serializeCDCMutations: {}]",
- dataTableName, encodedRegionName, batchSize, pollIntervalMs,
timestampBufferMs,
- startupDelayMs, pause, maxDataVisibilityRetries, parentProgressPauseMs,
+ + " [batchSize: {}, pollIntervalMs: {}, idlePollIntervalMs: {},
timestampBufferMs: {},"
+ + " startupDelayMs: {}, pause: {}, maxDataVisibilityRetries: {},"
+ + " parentProgressPauseMs: {}, serializeCDCMutations: {}]",
+ dataTableName, encodedRegionName, batchSize, pollIntervalMs,
idlePollIntervalMs,
+ timestampBufferMs, startupDelayMs, pause, maxDataVisibilityRetries,
parentProgressPauseMs,
serializeCDCMutations);
if (!waitForCDCStreamEntry()) {
LOG.error(
@@ -502,10 +514,10 @@ public class IndexCDCConsumer implements Runnable {
lastProcessedTimestamp =
processCDCBatchGenerated(encodedRegionName,
encodedRegionName, lastProcessedTimestamp, false);
}
+ retryCount = 0;
if (lastProcessedTimestamp == previousTimestamp) {
- sleepWithLagSampling(ConnectionUtils.getPauseTime(pause,
++retryCount));
+ sleepWithLagSampling(idlePollIntervalMs);
} else {
- retryCount = 0;
sleepWithLagSampling(pollIntervalMs);
}
} catch (Exception e) {
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexCDCConsumerLagIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexCDCConsumerLagIT.java
index 158c54d276..db2ff8871e 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexCDCConsumerLagIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexCDCConsumerLagIT.java
@@ -58,7 +58,6 @@ public class IndexCDCConsumerLagIT extends
ParallelStatsDisabledIT {
private static final int TIMESTAMP_BUFFER_MS = 2_000;
private static final int POLL_INTERVAL_MS = 500;
private static final int LAG_SAMPLE_INTERVAL_MS = 500;
- // Small retry pause so empty-poll backoff doesn't dominate idle behavior.
private static final int RETRY_PAUSE_MS = 100;
// Budget for the consumer to start up and emit its first lag sample.
Generous because the
// consumer waits up to INDEX_CDC_CONSUMER_STARTUP_DELAY_MS (default 10s)
and then performs
