Google Git
Sign in
apache/phoenix/refs/heads/tmp-ec^!/.
commit
cc964191b9f0d678182533b3288a295ed613bd49
[log]
author
Viraj Jasani <vjasani@apache.org>
Mon Apr 06 17:09:40 2026 -0700
committer
Viraj Jasani <vjasani@apache.org>
Mon Apr 06 17:10:49 2026 -0700
tree
e3704f6e3be368839c4264042b83179939f223d4
parent
37955ca27d68f0caa549e4b84a7694b99f3d1349 [diff]
better use of parentProgressPauseMs

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 1ed4e5f..9eecbe2 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

@@ -808,11 +808,16 @@
           }
           long otherProgress = getParentProgress(partitionId);
           if (otherProgress > currentLastProcessedTimestamp) {
-            sleepIfNotStopped(parentProgressPauseMs);
-            if (isPartitionCompleted(partitionId)) {
-              return;
-            }
-            currentLastProcessedTimestamp = getParentProgress(partitionId);
+            long previousOtherProgress;
+            do {
+              previousOtherProgress = otherProgress;
+              sleepIfNotStopped(parentProgressPauseMs);
+              if (isPartitionCompleted(partitionId)) {
+                return;
+              }
+              otherProgress = getParentProgress(partitionId);
+            } while (!stopped && otherProgress > previousOtherProgress);
+            currentLastProcessedTimestamp = otherProgress;
           }
         }
         long newTimestamp;

diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualGenerateIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualGenerateIT.java
index 1bf63c7..0088a38 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualGenerateIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsCoveredEventualGenerateIT.java

@@ -18,6 +18,7 @@
 package org.apache.phoenix.end2end;
 
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_BATCH_SIZE;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_RETRY_PAUSE_MS;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS;
 import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_MUTATION_SERIALIZE;
@@ -61,6 +62,7 @@
     props.put(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, Integer.toString(-1));
     props.put("hbase.coprocessor.master.classes", PhoenixMasterObserver.class.getName());
     props.put(PHOENIX_INDEX_CDC_MUTATION_SERIALIZE, Boolean.FALSE.toString());
+    props.put(INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS, Integer.toString(1000));
     setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
   }
 

diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
index bda94a3..4e10284 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java

@@ -19,6 +19,7 @@
 
 import static org.apache.phoenix.end2end.IndexToolIT.verifyIndexTable;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_BATCH_SIZE;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS;
 import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_MUTATION_SERIALIZE;
 import static org.junit.Assert.assertEquals;
@@ -100,6 +101,7 @@
     props.put(INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS, Integer.toString(1000));
     props.put("hbase.coprocessor.master.classes", PhoenixMasterObserver.class.getName());
     props.put(PHOENIX_INDEX_CDC_MUTATION_SERIALIZE, Boolean.TRUE.toString());
+    props.put(INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS, Integer.toString(1000));
     setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
   }
 

diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java
index 54ad81e..35fcbf1 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIndexIT.java

@@ -136,6 +136,7 @@
     splitThread.join(10000);
     LOGGER.info(
       "Total upsert time in ms : " + (EnvironmentEdgeManager.currentTimeMillis() - startTime));
+    Thread.sleep(20000);
     List<String> allIndexes =
       new ArrayList<>(Arrays.asList(indexName1, indexName2, indexName3, indexName4, indexName5));
     verifyRandomIndexes(allIndexes, schemaName, tableName, conn, nRows);
@@ -257,6 +258,7 @@
     assertTrue("Ran out of time", doneSignal.await(1500, TimeUnit.SECONDS));
     LOGGER.info(
       "Total upsert time in ms : " + (EnvironmentEdgeManager.currentTimeMillis() - startTime));
+    Thread.sleep(20000);
     List<String> allIndexes = new ArrayList<>(
       Arrays.asList(indexName1, indexName2, indexName3, indexName4, indexName5, indexName6));
     verifyRandomIndexes(allIndexes, schemaName, tableName, conn, nRows);

diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualGenerateIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualGenerateIT.java
index 5ecfc1c..24511b4 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualGenerateIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsUncoveredEventualGenerateIT.java

@@ -18,6 +18,7 @@
 package org.apache.phoenix.end2end;
 
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_BATCH_SIZE;
+import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_RETRY_PAUSE_MS;
 import static org.apache.phoenix.hbase.index.IndexCDCConsumer.INDEX_CDC_CONSUMER_TIMESTAMP_BUFFER_MS;
 import static org.apache.phoenix.hbase.index.IndexRegionObserver.PHOENIX_INDEX_CDC_MUTATION_SERIALIZE;
@@ -61,6 +62,7 @@
     props.put(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, Integer.toString(-1));
     props.put("hbase.coprocessor.master.classes", PhoenixMasterObserver.class.getName());
     props.put(PHOENIX_INDEX_CDC_MUTATION_SERIALIZE, Boolean.FALSE.toString());
+    props.put(INDEX_CDC_CONSUMER_PARENT_PROGRESS_PAUSE_MS, Integer.toString(1000));
     setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
   }