virajjasani commented on code in PR #2211:
URL: https://github.com/apache/phoenix/pull/2211#discussion_r2184359348
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/CDCStreamTTLIT.java:
##########
@@ -0,0 +1,72 @@
+package org.apache.phoenix.end2end;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessor.PhoenixMasterObserver;
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.util.Map;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class CDCStreamTTLIT extends CDCBaseIT {
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+
props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
Integer.toString(60*60)); // An hour
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION,
Boolean.toString(false));
+ props.put(QueryServices.TASK_HANDLING_INTERVAL_MS_ATTRIB,
+ Long.toString(Long.MAX_VALUE));
+ props.put(QueryServices.TASK_HANDLING_INITIAL_DELAY_MS_ATTRIB,
+ Long.toString(Long.MAX_VALUE));
+
props.put(QueryServices.PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS,
Long.toString(10*1000));
+ props.put("hbase.coprocessor.master.classes",
PhoenixMasterObserver.class.getName());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ taskRegionEnvironment =
+ getUtility()
+ .getRSForFirstRegionInTable(
+
PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME)
+
.getRegions(PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME)
+ .get(0).getCoprocessorHost()
+
.findCoprocessorEnvironment(TaskRegionObserver.class.getName());
+ }
+
+ @Test
+ public void testCDCStreamTTL() throws Exception {
+ Connection conn = newConnection();
+ String tableName = generateUniqueName();
+ createTableAndEnableCDC(conn, tableName, true);
+ TestUtil.splitTable(conn, tableName, Bytes.toBytes("m"));
+ String sql = "SELECT PARTITION_END_TIME FROM SYSTEM.CDC_STREAM WHERE
TABLE_NAME='" + tableName + "'";
+ ResultSet rs = conn.createStatement().executeQuery(sql);
+ int count = 0;
+ while (rs.next()) {
+ count++;
+ }
+ Assert.assertEquals(3, count);
+ Thread.sleep(11000);
Review Comment:
Instead of sleep, using `ManualEnvironmentEdge` would be much better
approach.
Even if we don't configure `PHOENIX_CDC_STREAM_PARTITION_EXPIRY_MIN_AGE_MS`
for the test, we can still increment the clock manually by more than 30 hr and
without sleep, we can verify that the parent closed partition is no longer
visible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
