From 445abb979db324b23997ac14f12c51a04a714e8a Mon Sep 17 00:00:00 2001
From: Palash Chauhan <p.chauhan@salesforce.com>
Date: Wed, 12 Aug 2026 10:05:10 -0700
Subject: [PATCH] PHOENIX-7986 : Replace IndexCDCConsumer's exponential idle
 backoff with a fixed poll interval

---
 .../phoenix/hbase/index/IndexCDCConsumer.java | 26 ++++++++++++++-----
 .../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 25b0e18f268..46423f0d87f 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 IndexCDCConsumer(RegionCoprocessorEnvironment env, String dataTableName,
     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 void run() {
       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 void run() {
               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 158c54d2767..db2ff8871e8 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