Copilot commented on code in PR #19265:
URL: https://github.com/apache/pinot/pull/19265#discussion_r3790116599
##########
pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java:
##########
@@ -979,19 +990,26 @@ protected long getCurrentCountStarResult(String
tableName) {
}
protected void waitForMinionTaskCompletion(String taskId, long timeout) {
- TestUtils.waitForCondition(aVoid ->
-
_controllerStarter.getHelixTaskResourceManager().getTaskState(taskId) ==
TaskState.COMPLETED,
- timeout, "Failed to complete the task " + taskId);
-
- // Validate that there were > 0 subtasks so that we know the task was
actually run
-
Assert.assertFalse(_controllerStarter.getHelixTaskResourceManager().getSubtaskStates(taskId).isEmpty());
+ // The task state (workflow context) and the subtask states (job context)
live in different Helix znodes and are
+ // not updated atomically, so a subtask can still read as RUNNING (or
null, not yet started) right after the task
+ // turns COMPLETED. Wait until the task is COMPLETED and every subtask
reached a terminal state before validating
+ // them. A non-empty subtask map also proves the task was actually run.
+ PinotHelixTaskResourceManager taskResourceManager =
_controllerStarter.getHelixTaskResourceManager();
+ TestUtils.waitForCondition(aVoid -> {
+ if (taskResourceManager.getTaskState(taskId) != TaskState.COMPLETED) {
+ return false;
+ }
+ Map<String, TaskPartitionState> subtaskStates =
taskResourceManager.getSubtaskStates(taskId);
+ return !subtaskStates.isEmpty() && subtaskStates.values()
+ .stream()
+ .noneMatch(state -> state == null || state ==
TaskPartitionState.INIT || state == TaskPartitionState.RUNNING);
Review Comment:
`STOPPED` is also non-terminal for Helix task partitions because a stopped
partition can be resumed. Treating every state other than `null`, `INIT`, and
`RUNNING` as terminal lets a stale `STOPPED` value satisfy this wait, after
which the unchanged assertion can fail immediately—the same race this change is
intended to remove. Include `STOPPED` in the states that keep polling.
##########
pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java:
##########
@@ -329,22 +333,20 @@ protected Schema createSchema(File schemaFile)
protected TableConfig createTableConfig(String tableConfigFileName)
throws IOException {
URL configPathUrl =
getClass().getClassLoader().getResource(tableConfigFileName);
- Assert.assertNotNull(configPathUrl);
+ assertNotNull(configPathUrl);
return createTableConfig(new File(configPathUrl.getFile()));
}
protected TableConfig createTableConfig(File tableConfigFile)
throws IOException {
InputStream inputStream = new FileInputStream(tableConfigFile);
- Assert.assertNotNull(inputStream);
+ assertNotNull(inputStream);
return JsonUtils.inputStreamToObject(inputStream, TableConfig.class);
}
/// Creates a new OFFLINE table config.
protected TableConfig createOfflineTableConfig() {
- // @formatter:off
- return new TableConfigBuilder(TableType.OFFLINE)
- .setTableName(getTableName())
+ return new
TableConfigBuilder(TableType.OFFLINE).setTableName(getTableName())
Review Comment:
This begins a series of formatting-only changes across unrelated builders,
Kafka helpers, connection setup, and assertions, while the PR's behavior change
is confined to the minion wait helper. The repository requires minimal diffs
and preserving unrelated formatting; please revert these non-functional hunks
so the race fix remains reviewable and avoids unnecessary merge conflicts.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
