You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2026/03/19 17:48:37 UTC
(pinot) branch master updated: [Testing] Always raise error when condition is not met (#17909)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 716de7f6be0 [Testing] Always raise error when condition is not met (#17909)
716de7f6be0 is described below
commit 716de7f6be0a8aed995f921f4dd01f80174bc1ea
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Thu Mar 19 10:48:30 2026 -0700
[Testing] Always raise error when condition is not met (#17909)
---
.../test/java/org/apache/pinot/util/TestUtils.java | 19 ++++++----
.../java/org/apache/pinot/compat/StreamOp.java | 3 +-
.../tests/BaseClusterIntegrationTest.java | 40 ++++++++++------------
.../tests/BasePauselessRealtimeIngestionTest.java | 4 +--
.../integration/tests/CursorIntegrationTest.java | 2 +-
.../GroupByEnableTrimOptionIntegrationTest.java | 3 +-
.../tests/LLCRealtimeClusterIntegrationTest.java | 3 +-
.../tests/MultiStageEngineIntegrationTest.java | 7 ++--
...pRealtimeIngestionSegmentCommitFailureTest.java | 2 +-
...sRealtimeIngestionSegmentCommitFailureTest.java | 2 +-
.../tests/PurgeMinionClusterIntegrationTest.java | 2 +-
...fflineSegmentsMinionClusterIntegrationTest.java | 3 +-
.../tests/RetentionManagerIntegrationTest.java | 2 +-
.../tests/RowLevelSecurityIntegrationTest.java | 5 +--
.../integration/tests/SSBQueryIntegrationTest.java | 5 +--
...mentGenerationMinionClusterIntegrationTest.java | 4 +--
...gmentGenerationMinionRealtimeIngestionTest.java | 4 +--
.../tests/SegmentUploadIntegrationTest.java | 17 ++-------
.../SegmentWriterUploaderIntegrationTest.java | 8 ++---
.../SparkSegmentMetadataPushIntegrationTest.java | 11 ++----
.../UpsertCompactMergeTaskIntegrationTest.java | 18 +++-------
.../BaseLogicalTableIntegrationTest.java | 2 +-
.../ingestion/BaseKinesisIntegrationTest.java | 20 +++++------
.../ingestion/RealtimeKinesisIntegrationTest.java | 11 ++----
24 files changed, 80 insertions(+), 117 deletions(-)
diff --git a/pinot-common/src/test/java/org/apache/pinot/util/TestUtils.java b/pinot-common/src/test/java/org/apache/pinot/util/TestUtils.java
index 2bbf4b495a5..b1d5dcc5803 100644
--- a/pinot-common/src/test/java/org/apache/pinot/util/TestUtils.java
+++ b/pinot-common/src/test/java/org/apache/pinot/util/TestUtils.java
@@ -94,6 +94,8 @@ public class TestUtils {
waitForCondition(condition, 100L, timeoutMs, errorMessage);
}
+ /// @deprecated Always raise error when condition is not met.
+ @Deprecated
public static void waitForCondition(Function<Void, Boolean> condition, long checkIntervalMs, long timeoutMs,
@Nullable String errorMessage, boolean raiseError) {
long endTime = System.currentTimeMillis() + timeoutMs;
@@ -113,12 +115,15 @@ public class TestUtils {
}
}
- /**
- * Like {@link #waitForCondition(Function, long, long, String, boolean)}, but supports functions that throw checked
- * exceptions.
- *
- * The first time every logPeriod that the function throws an exception, it is printed in the standard error.
- */
+ /// Like [#waitForCondition(Function, long, long, String)], but supports functions that throw checked exceptions.
+ /// The first time every logPeriod that the function throws an exception, it is logged as warning.
+ public static void waitForCondition(SupplierWithException<Boolean> condition, long checkIntervalMs, long timeoutMs,
+ @Nullable String errorMessage, @Nullable Duration logPeriod) {
+ waitForCondition(condition, checkIntervalMs, timeoutMs, errorMessage, true, logPeriod);
+ }
+
+ /// @deprecated Always raise error when condition is not met.
+ @Deprecated
public static void waitForCondition(SupplierWithException<Boolean> condition, long checkIntervalMs, long timeoutMs,
@Nullable String errorMessage, boolean raiseError, @Nullable Duration logPeriod) {
long endTime = System.currentTimeMillis() + timeoutMs;
@@ -137,8 +142,8 @@ public class TestUtils {
break;
} catch (Exception e) {
lastError = e;
+ numErrors++;
if (logPeriod != null) {
- numErrors++;
Instant now = Instant.now();
if (logPeriod.compareTo(Duration.between(errorLogInstant, now)) < 0) {
LOGGER.warn("Error while waiting for condition", e);
diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
index 12eaee7360b..626b2d14fbe 100644
--- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
+++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/StreamOp.java
@@ -311,8 +311,7 @@ public class StreamOp extends BaseOp {
long existingTotalDocs = fetchExistingTotalDocs(tableName);
loadedDocs.set(existingTotalDocs);
return existingTotalDocs == targetDocs;
- }, 100L, timeoutMs,
- "Failed to load " + targetDocs + " documents. Found " + loadedDocs.get() + " instead", true,
+ }, 100L, timeoutMs, "Failed to load " + targetDocs + " documents. Found " + loadedDocs.get() + " instead",
Duration.ofSeconds(1));
}
}
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index 56e62dee28d..6b41fe3ab2f 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -97,8 +97,6 @@ import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.tools.utils.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
import org.intellij.lang.annotations.Language;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.Assert;
@@ -106,8 +104,6 @@ import org.testng.Assert;
* Shared implementation details of the cluster integration tests.
*/
public abstract class BaseClusterIntegrationTest extends ClusterTest {
- private static final Logger LOGGER = LoggerFactory.getLogger(BaseClusterIntegrationTest.class);
-
// Default settings
protected static final String DEFAULT_TABLE_NAME = "mytable";
protected static final String DEFAULT_LOGICAL_TABLE_NAME = "mytable_logical";
@@ -781,7 +777,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableConfig.getTableName()) == expectedNoOfDocs, 100L,
timeoutMs, "Failed to load " + expectedNoOfDocs + " documents in table " + tableConfig.getTableName(),
- true, Duration.ofMillis(timeoutMs / 10));
+ Duration.ofMillis(timeoutMs / 10));
}
protected List<File> getAllAvroFiles()
@@ -1062,9 +1058,22 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
*/
protected void waitForAllDocsLoaded(long timeoutMs)
throws Exception {
- waitForDocsLoaded(timeoutMs, true, getTableName());
+ waitForAllDocsLoaded(getTableName(), timeoutMs);
+ }
+
+ protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
+ long countStarResult = getCountStarResult();
+ TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) == countStarResult, 100L, timeoutMs,
+ "Failed to load " + countStarResult + " documents", Duration.ofMillis(timeoutMs / 10));
+ }
+
+ protected void waitForAnyDocLoaded(String tableName, long timeoutMs) {
+ TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) > 0, 100L, timeoutMs,
+ "Failed to load any document", Duration.ofMillis(timeoutMs / 10));
}
+ /// @deprecated Always raise error when condition is not met.
+ @Deprecated
protected void waitForDocsLoaded(long timeoutMs, boolean raiseError, String tableName) {
long countStarResult = getCountStarResult();
TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) == countStarResult, 100L, timeoutMs,
@@ -1073,16 +1082,10 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
protected void waitForAllRealtimePartitionsConsuming(String tableNameWithType, long timeoutMs) {
int expectedPartitions = getNumKafkaPartitions();
- TestUtils.waitForCondition(() -> hasConsumingSegmentsForAllPartitions(tableNameWithType, expectedPartitions), 200L,
- timeoutMs, "Failed to get CONSUMING segments for all " + expectedPartitions + " partitions for table "
- + tableNameWithType, false, Duration.ofMillis(timeoutMs / 10));
- int consumingPartitions = getNumConsumingPartitions(tableNameWithType);
- Assert.assertEquals(consumingPartitions, expectedPartitions, "Expected CONSUMING segments for "
- + expectedPartitions + " partitions, found " + consumingPartitions + " for table " + tableNameWithType);
- }
-
- private boolean hasConsumingSegmentsForAllPartitions(String tableNameWithType, int expectedPartitions) {
- return getNumConsumingPartitions(tableNameWithType) >= expectedPartitions;
+ TestUtils.waitForCondition(() -> getNumConsumingPartitions(tableNameWithType) == expectedPartitions, 200L,
+ timeoutMs,
+ "Failed to get CONSUMING segments for all " + expectedPartitions + " partitions for table " + tableNameWithType,
+ Duration.ofMillis(timeoutMs / 10));
}
private int getNumConsumingPartitions(String tableNameWithType) {
@@ -1104,11 +1107,6 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
return consumingPartitions.size();
}
- protected void waitForNonZeroDocsLoaded(long timeoutMs, boolean raiseError, String tableName) {
- TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) > 0, 100L, timeoutMs,
- "Failed to load non zero documents", raiseError, Duration.ofMillis(timeoutMs / 10));
- }
-
/**
* Wait for servers to remove the table data manager after the table is deleted.
*/
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasePauselessRealtimeIngestionTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasePauselessRealtimeIngestionTest.java
index 9c3a933c27e..8d93d5fea1d 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasePauselessRealtimeIngestionTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasePauselessRealtimeIngestionTest.java
@@ -126,7 +126,7 @@ public abstract class BasePauselessRealtimeIngestionTest extends BaseClusterInte
tableConfig.getValidationConfig().setRetentionTimeValue("100000");
addTableConfig(tableConfig);
- waitForDocsLoaded(600_000L, true, tableConfig.getTableName());
+ waitForAllDocsLoaded(tableConfig.getTableName(), 600_000L);
TestUtils.waitForCondition((aVoid) -> {
List<SegmentZKMetadata> segmentZKMetadataList =
_helixResourceManager.getSegmentsZKMetadata(tableConfig.getTableName());
@@ -214,7 +214,7 @@ public abstract class BasePauselessRealtimeIngestionTest extends BaseClusterInte
_controllerStarter.getRealtimeSegmentValidationManager().run();
waitForAllDocsLoaded(600_000L);
- waitForDocsLoaded(600_000L, true, tableNameWithType2);
+ waitForAllDocsLoaded(tableNameWithType2, 600_000L);
PauselessRealtimeTestUtils.verifyIdealState(tableNameWithType, NUM_REALTIME_SEGMENTS, _helixManager);
PauselessRealtimeTestUtils.verifyIdealState(tableNameWithType2, NUM_REALTIME_SEGMENTS, _helixManager);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorIntegrationTest.java
index d7d4b3488e1..64295ea781e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/CursorIntegrationTest.java
@@ -420,6 +420,6 @@ public class CursorIntegrationTest extends BaseClusterIntegrationTestSet {
LOGGER.error(e.getMessage());
return false;
}
- }, 500L, 100_000L, "Failed to load delete query results", true);
+ }, 500L, 100_000L, "Failed to load delete query results");
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/GroupByEnableTrimOptionIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/GroupByEnableTrimOptionIntegrationTest.java
index 98027bc7d66..7df8f6f0327 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/GroupByEnableTrimOptionIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/GroupByEnableTrimOptionIntegrationTest.java
@@ -76,8 +76,7 @@ public class GroupByEnableTrimOptionIntegrationTest extends BaseClusterIntegrati
// Wait for all documents loaded
TestUtils.waitForCondition(() -> getCurrentCountStarResult(DEFAULT_TABLE_NAME) == FILES_NO * RECORDS_NO, 100L,
- 60_000,
- "Failed to load documents", true, Duration.ofMillis(60_000 / 10));
+ 60_000L, "Failed to load documents", Duration.ofMillis(60_000 / 10));
setUseMultiStageQueryEngine(true);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
index 49838cf2148..4673ce50661 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
@@ -140,8 +140,7 @@ public class LLCRealtimeClusterIntegrationTest extends BaseRealtimeClusterIntegr
continue;
}
TestUtils.waitForCondition(() -> isOffline(partition, seqNum), 5000L, timeoutMs,
- "Failed to find offline segment in partition " + partition + " seqNum ", true,
- Duration.ofMillis(timeoutMs / 10));
+ "Failed to find offline segment in partition " + partition + " seqNum ", Duration.ofMillis(timeoutMs / 10));
getControllerRequestClient().runPeriodicTask("RealtimeSegmentValidationManager");
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
index db47b3997c7..1248d20dd66 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
@@ -78,7 +78,10 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.apache.pinot.common.function.scalar.StringFunctions.*;
-import static org.testng.Assert.*;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
public class MultiStageEngineIntegrationTest extends BaseClusterIntegrationTestSet {
@@ -2255,7 +2258,7 @@ public class MultiStageEngineIntegrationTest extends BaseClusterIntegrationTestS
Assert.assertNull(mytableLeaf.get("children"), "When pipeline breaker stats are not kept, "
+ "there should be no children under the leaf node");
return true;
- }, 100, 10_000L, errorMsg, true, Duration.ofSeconds(1));
+ }, 100, 10_000L, errorMsg, Duration.ofSeconds(1));
}
@AfterClass
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupRealtimeIngestionSegmentCommitFailureTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupRealtimeIngestionSegmentCommitFailureTest.java
index bba8191954a..a3b04b3db0b 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupRealtimeIngestionSegmentCommitFailureTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupRealtimeIngestionSegmentCommitFailureTest.java
@@ -124,7 +124,7 @@ public class PauselessDedupRealtimeIngestionSegmentCommitFailureTest
tableConfig2.getValidationConfig().setRetentionTimeUnit("DAYS");
tableConfig2.getValidationConfig().setRetentionTimeValue("100000");
addTableConfig(tableConfig2);
- waitForDocsLoaded(600_000L, true, tableConfig2.getTableName());
+ waitForAllDocsLoaded(tableConfig2.getTableName(), 600_000L);
// create schema for pauseless table
schema.setSchemaName(getPauselessTableName());
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessRealtimeIngestionSegmentCommitFailureTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessRealtimeIngestionSegmentCommitFailureTest.java
index 23c9bcc086d..5c578d20b5e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessRealtimeIngestionSegmentCommitFailureTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessRealtimeIngestionSegmentCommitFailureTest.java
@@ -116,7 +116,7 @@ public class PauselessRealtimeIngestionSegmentCommitFailureTest extends BaseClus
tableConfig2.getValidationConfig().setRetentionTimeValue("100000");
addTableConfig(tableConfig2);
- waitForDocsLoaded(600_000L, true, tableConfig2.getTableName());
+ waitForAllDocsLoaded(tableConfig2.getTableName(), 600_000L);
// create schema for pauseless table
schema.setSchemaName(getPauselessTableName());
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java
index 10d69b04ba2..1ac32032960 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java
@@ -129,7 +129,7 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes
// Push data into Kafka to create LLC segments
pushAvroIntoKafka(avroFiles);
// Wait for all documents loaded
- waitForDocsLoaded(600_000L, true, PURGE_REALTIME_LAST_SEGMENT_TABLE);
+ waitForAllDocsLoaded(PURGE_REALTIME_LAST_SEGMENT_TABLE, 600_000L);
setRecordPurger();
_helixTaskResourceManager = _controllerStarter.getHelixTaskResourceManager();
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
index 3d8cb711696..1ab44e6500e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
@@ -162,8 +162,7 @@ public class RealtimeToOfflineSegmentsMinionClusterIntegrationTest extends BaseC
// Wait for all documents loaded
waitForAllDocsLoaded(600_000L);
-
- waitForDocsLoaded(600_000L, true, tableWithMetadataPush);
+ waitForAllDocsLoaded(tableWithMetadataPush, 600_000L);
_taskResourceManager = _controllerStarter.getHelixTaskResourceManager();
_taskManager = _controllerStarter.getTaskManager();
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RetentionManagerIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RetentionManagerIntegrationTest.java
index cc14afc1b5a..7131eb37b03 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RetentionManagerIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RetentionManagerIntegrationTest.java
@@ -94,7 +94,7 @@ public class RetentionManagerIntegrationTest extends BaseClusterIntegrationTest
tableConfig.getValidationConfig().setRetentionTimeValue("2");
addTableConfig(tableConfig);
- waitForDocsLoaded(600_000L, true, tableConfig.getTableName());
+ waitForAllDocsLoaded(600_000L);
}
@AfterClass
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
index 65ea77da328..9a23657983e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RowLevelSecurityIntegrationTest.java
@@ -139,13 +139,10 @@ public class RowLevelSecurityIntegrationTest extends BaseClusterIntegrationTest
setupTable(DEFAULT_TABLE_NAME);
setupTable(DEFAULT_TABLE_NAME_2);
setupTable(DEFAULT_TABLE_NAME_3);
-
- waitForAllDocsLoaded(600_000L);
}
private void setupTable(String tableName)
throws Exception {
-
Schema schema = createSchema();
schema.setSchemaName(tableName);
addSchema(schema);
@@ -156,7 +153,7 @@ public class RowLevelSecurityIntegrationTest extends BaseClusterIntegrationTest
tableConfig.getValidationConfig().setRetentionTimeValue("100000");
addTableConfig(tableConfig);
- waitForDocsLoaded(600_000L, true, tableConfig.getTableName());
+ waitForAllDocsLoaded(tableName, 600_000L);
}
@AfterClass
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SSBQueryIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SSBQueryIntegrationTest.java
index 15d1baf2e5a..3b509d64dcb 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SSBQueryIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SSBQueryIntegrationTest.java
@@ -32,8 +32,6 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.tools.utils.JarUtils;
import org.apache.pinot.util.TestUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -43,7 +41,6 @@ import org.yaml.snakeyaml.Yaml;
public class SSBQueryIntegrationTest extends BaseClusterIntegrationTest {
- private static final Logger LOGGER = LoggerFactory.getLogger(SSBQueryIntegrationTest.class);
private static final Map<String, String> SSB_QUICKSTART_TABLE_RESOURCES = Map.of(
"customer", "examples/batch/ssb/customer",
"dates", "examples/batch/ssb/dates",
@@ -93,7 +90,7 @@ public class SSBQueryIntegrationTest extends BaseClusterIntegrationTest {
uploadSegments(tableName, _tarDir);
// H2
ClusterIntegrationTestUtils.setUpH2TableWithAvro(Collections.singletonList(dataFile), tableName, _h2Connection);
- waitForNonZeroDocsLoaded(60_000L, true, tableName);
+ waitForAnyDocLoaded(tableName, 60_000L);
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionClusterIntegrationTest.java
index bd4673f8536..561ce8e7734 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionClusterIntegrationTest.java
@@ -101,7 +101,7 @@ public class SegmentGenerationMinionClusterIntegrationTest extends BaseClusterIn
LOGGER.error("Failed to get expected totalDocs: {}", rowCnt, e);
return false;
}
- }, 5000L, 600_000L, "Failed to load " + rowCnt + " documents", true);
+ }, 5000L, 600_000L, "Failed to load " + rowCnt + " documents");
JsonNode result = postQuery("SELECT COUNT(*) FROM " + tableName);
// One segment per file.
assertEquals(result.get("numSegmentsQueried").asInt(), 7);
@@ -146,7 +146,7 @@ public class SegmentGenerationMinionClusterIntegrationTest extends BaseClusterIn
LOGGER.error("Failed to get expected totalDocs: {}", rowCnt, e);
return false;
}
- }, 5000L, 600_000L, "Failed to load " + rowCnt + " documents", true);
+ }, 5000L, 600_000L, "Failed to load " + rowCnt + " documents");
JsonNode result = postQuery("SELECT COUNT(*) FROM " + tableName);
// One segment per file.
assertEquals(result.get("numSegmentsQueried").asInt(), 7);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionRealtimeIngestionTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionRealtimeIngestionTest.java
index 40bb89d12d7..3aa5dfb1a8d 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionRealtimeIngestionTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentGenerationMinionRealtimeIngestionTest.java
@@ -117,7 +117,7 @@ public class SegmentGenerationMinionRealtimeIngestionTest extends BaseClusterInt
} catch (Exception e) {
return false;
}
- }, 5000L, 600_000L, "Failed to load " + DEFAULT_COUNT_STAR_RESULT + " documents", true);
+ }, 5000L, 600_000L, "Failed to load " + DEFAULT_COUNT_STAR_RESULT + " documents");
JsonNode result = postQuery("SELECT COUNT(*) FROM " + REALTIME_TABLE_NAME);
assertEquals(result.get("numSegmentsQueried").asInt(), 14);
}
@@ -152,7 +152,7 @@ public class SegmentGenerationMinionRealtimeIngestionTest extends BaseClusterInt
} catch (Exception e) {
return false;
}
- }, 5000L, 600_000L, "Failed to load " + DEFAULT_COUNT_STAR_RESULT + " documents", true);
+ }, 5000L, 600_000L, "Failed to load " + DEFAULT_COUNT_STAR_RESULT + " documents");
JsonNode result = postQuery("SELECT COUNT(*) FROM " + REALTIME_TABLE_NAME);
assertEquals(result.get("numSegmentsQueried").asInt(), 14);
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
index bbad2496802..b4a414b8f1f 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java
@@ -19,13 +19,11 @@
package org.apache.pinot.integration.tests;
import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.base.Function;
import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
-import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.pinot.controller.helix.ControllerTest;
@@ -422,18 +420,9 @@ public class SegmentUploadIntegrationTest extends BaseClusterIntegrationTest {
.get(0).get("OFFLINE");
}
- protected void testCountStar(final long countStarResult) {
- TestUtils.waitForCondition(new Function<Void, Boolean>() {
- @Nullable
- @Override
- public Boolean apply(@Nullable Void aVoid) {
- try {
- return getCurrentCountStarResult() == countStarResult;
- } catch (Exception e) {
- return null;
- }
- }
- }, 100L, 300_000, "Failed to load " + countStarResult + " documents", true);
+ protected void testCountStar(long countStarResult) {
+ TestUtils.waitForCondition(() -> getCurrentCountStarResult() == countStarResult, 100L, 300_000L,
+ "Failed to load " + countStarResult + " documents", null);
}
@Override
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java
index c0ad63c550f..eeec1634bd0 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentWriterUploaderIntegrationTest.java
@@ -181,7 +181,7 @@ public class SegmentWriterUploaderIntegrationTest extends BaseClusterIntegration
}
private void checkTotalDocsInQuery(long expectedTotalDocs) {
- TestUtils.waitForCondition(new Function<Void, Boolean>() {
+ TestUtils.waitForCondition(new Function<>() {
@Nullable
@Override
public Boolean apply(@Nullable Void aVoid) {
@@ -192,11 +192,11 @@ public class SegmentWriterUploaderIntegrationTest extends BaseClusterIntegration
return null;
}
}
- }, 100L, 120_000, "Failed to load " + expectedTotalDocs + " documents", true);
+ }, 100L, 120_000L, "Failed to load " + expectedTotalDocs + " documents");
}
private void checkNumSegments(int expectedNumSegments) {
- TestUtils.waitForCondition(new Function<Void, Boolean>() {
+ TestUtils.waitForCondition(new Function<>() {
@Nullable
@Override
public Boolean apply(@Nullable Void aVoid) {
@@ -207,7 +207,7 @@ public class SegmentWriterUploaderIntegrationTest extends BaseClusterIntegration
return null;
}
}
- }, 100L, 120_000, "Failed to load get num segments", true);
+ }, 100L, 120_000L, "Failed to load get num segments");
}
@AfterClass
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SparkSegmentMetadataPushIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SparkSegmentMetadataPushIntegrationTest.java
index aacb18530aa..b6b4eb28870 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SparkSegmentMetadataPushIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SparkSegmentMetadataPushIntegrationTest.java
@@ -369,14 +369,9 @@ public class SparkSegmentMetadataPushIntegrationTest extends BaseClusterIntegrat
.get("OFFLINE");
}
- protected void testCountStar(final long countStarResult) {
- TestUtils.waitForCondition(aVoid -> {
- try {
- return getCurrentCountStarResult() == countStarResult;
- } catch (Exception e) {
- return null;
- }
- }, 100L, 300_000, "Failed to load " + countStarResult + " documents", true);
+ protected void testCountStar(long countStarResult) {
+ TestUtils.waitForCondition(() -> getCurrentCountStarResult() == countStarResult, 100L, 300_000L,
+ "Failed to load " + countStarResult + " documents", null);
}
@AfterMethod
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertCompactMergeTaskIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertCompactMergeTaskIntegrationTest.java
index 6b04c8ea467..e0198ebf3d6 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertCompactMergeTaskIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertCompactMergeTaskIntegrationTest.java
@@ -110,7 +110,8 @@ public class UpsertCompactMergeTaskIntegrationTest extends BaseClusterIntegratio
setUpTable(getTableName(), kafkaTopicName, null);
// Wait for all documents loaded
- waitForTotalDocsLoaded(600_000L, 10);
+ TestUtils.waitForCondition(() -> getCurrentCountStarResultSkipUpsert() == 10, 100L, 600_000L,
+ "Failed to load 10 documents", Duration.ofMillis(600_000L / 10));
assertEquals(getCurrentCountStarResult(), getCountStarResult());
// Create partial upsert table schema
@@ -710,20 +711,9 @@ public class UpsertCompactMergeTaskIntegrationTest extends BaseClusterIntegratio
return null;
}
- protected void waitForTotalDocsLoaded(long timeoutMs, int totalDoc)
- throws Exception {
- waitForDocsLoaded(timeoutMs, true, getTableName(), totalDoc);
- }
-
- protected void waitForDocsLoaded(long timeoutMs, boolean raiseError, String tableName, int totalDoc) {
- long countStarResult = getCountStarResult();
- TestUtils.waitForCondition(() -> getCurrentCountStarResultAll(tableName) == totalDoc, 100L, timeoutMs,
- "Failed to load " + countStarResult + " documents", raiseError, Duration.ofMillis(timeoutMs / 10));
- }
-
- protected long getCurrentCountStarResultAll(String tableName) {
+ private long getCurrentCountStarResultSkipUpsert() {
ResultSetGroup resultSetGroup =
- getPinotConnection().execute("SELECT COUNT(*) FROM " + tableName + " OPTION(skipUpsert=true)");
+ getPinotConnection().execute("SET skipUpsert = true; SELECT COUNT(*) FROM " + getTableName());
if (resultSetGroup.getResultSetCount() > 0) {
return resultSetGroup.getResultSet(0).getLong(0);
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
index 877e1b25489..b5d6ba5824f 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/BaseLogicalTableIntegrationTest.java
@@ -431,7 +431,7 @@ public abstract class BaseLogicalTableIntegrationTest extends BaseClusterIntegra
@Override
protected void waitForAllDocsLoaded(long timeoutMs)
throws Exception {
- waitForDocsLoaded(timeoutMs, true, getLogicalTableName());
+ waitForAllDocsLoaded(getLogicalTableName(), timeoutMs);
}
@Override
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/BaseKinesisIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/BaseKinesisIntegrationTest.java
index 13679c62252..80b131121a6 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/BaseKinesisIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/BaseKinesisIntegrationTest.java
@@ -116,9 +116,8 @@ abstract class BaseKinesisIntegrationTest extends BaseClusterIntegrationTest {
LOGGER.warn("Stream " + STREAM_NAME + " being created");
_kinesisClient.createStream(CreateStreamRequest.builder().streamName(STREAM_NAME).shardCount(numShards).build());
- TestUtils.waitForCondition(aVoid ->
- KinesisUtils.isKinesisStreamActive(_kinesisClient, STREAM_NAME), 2000L, 60000,
- "Kinesis stream " + STREAM_NAME + " is not created or is not in active state", true);
+ TestUtils.waitForCondition(aVoid -> KinesisUtils.isKinesisStreamActive(_kinesisClient, STREAM_NAME), 2000L, 60_000L,
+ "Kinesis stream " + STREAM_NAME + " is not created or is not in active state");
}
protected void deleteStream() {
@@ -128,14 +127,13 @@ abstract class BaseKinesisIntegrationTest extends BaseClusterIntegrationTest {
return;
}
TestUtils.waitForCondition(aVoid -> {
- try {
- KinesisUtils.getKinesisStreamStatus(_kinesisClient, STREAM_NAME);
- } catch (ResourceNotFoundException e) {
- return true;
- }
- return false;
- }, 2000L, 60000,
- "Kinesis stream " + STREAM_NAME + " is not deleted", true);
+ try {
+ KinesisUtils.getKinesisStreamStatus(_kinesisClient, STREAM_NAME);
+ return false;
+ } catch (ResourceNotFoundException e) {
+ return true;
+ }
+ }, 2000L, 60_000L, "Kinesis stream " + STREAM_NAME + " is not deleted");
LOGGER.warn("Stream " + STREAM_NAME + " deleted");
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/RealtimeKinesisIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/RealtimeKinesisIntegrationTest.java
index 666b78fb341..87f1f2b4a6f 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/RealtimeKinesisIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/ingestion/RealtimeKinesisIntegrationTest.java
@@ -84,13 +84,8 @@ public class RealtimeKinesisIntegrationTest extends BaseKinesisIntegrationTest {
waitForAllDocsLoadedKinesis(120_000L);
}
- protected void waitForAllDocsLoadedKinesis(long timeoutMs)
- throws Exception {
- waitForAllDocsLoadedKinesis(timeoutMs, true);
- }
-
- protected void waitForAllDocsLoadedKinesis(long timeoutMs, boolean raiseError) {
- TestUtils.waitForCondition(new Function<Void, Boolean>() {
+ protected void waitForAllDocsLoadedKinesis(long timeoutMs) {
+ TestUtils.waitForCondition(new Function<>() {
@Nullable
@Override
public Boolean apply(@Nullable Void aVoid) {
@@ -101,7 +96,7 @@ public class RealtimeKinesisIntegrationTest extends BaseKinesisIntegrationTest {
return null;
}
}
- }, 1000L, timeoutMs, "Failed to load " + _totalRecordsPushedInStream + " documents", raiseError);
+ }, 1000L, timeoutMs, "Failed to load " + _totalRecordsPushedInStream + " documents");
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org