This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new b2721b2cd1c fix: avoid race in
IngestionSmokeTest.test_streamLogs_ofCancelledTask (#20549)
b2721b2cd1c is described below
commit b2721b2cd1c76bcdffac6f712950c9db9e65ea83
Author: rishi-rana <[email protected]>
AuthorDate: Sat Oct 10 21:25:49 2026 -0700
fix: avoid race in IngestionSmokeTest.test_streamLogs_ofCancelledTask
(#20549)
* fix: avoid race in IngestionSmokeTest.test_streamLogs_ofCancelledTask
The peon pushes its task log after the process exits, so
TaskLogStreamer#streamTaskLog can return a present-but-empty stream before
the expected content is written. Poll until the log contains the expected
line instead of just until the stream is present.
Fixes #20435.
Assisted with AI Claude.
* fix: correct comment on IngestionSmokeTest log-race fix
@ykisana pointed out this test's setup has no peon — EmbeddedIndexer runs
tasks in-process and the log is served over HTTP from its own process log
file (HttpRemoteTaskRunner#streamTaskLog). Fix the comment to describe that
mechanism instead of a peon process-exit push.
Assisted with AI Claude.
* fix: close task-log stream on each poll in IngestionSmokeTest
Each retry of the waitForResult poll opened a new InputStream from
streamTaskLog without closing it. Close it with try-with-resources after
reading, so retries don't leak file/HTTP response descriptors.
Assisted with AI Claude.
---
.../embedded/indexing/IngestionSmokeTest.java | 32 ++++++++++++++--------
1 file changed, 20 insertions(+), 12 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
index b05c4731fb6..dcab6cabad8 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
@@ -429,22 +429,30 @@ public class IngestionSmokeTest extends
EmbeddedClusterTestBase
.hasDimension(DruidMetrics.TASK_STATUS, "FAILED")
);
- final Optional<InputStream> streamOptional =
- cluster.callApi().waitForResult(
- () -> overlord.bindings()
- .getInstance(TaskLogStreamer.class)
- .streamTaskLog(taskId, 0),
- Optional::isPresent
- ).go();
-
- Assertions.assertTrue(streamOptional.isPresent());
-
- final String logs = IOUtils.toString(streamOptional.get(),
StandardCharsets.UTF_8);
-
final String expectedLogLine = StringUtils.format(
"Running task[%s] for [%d] millis",
taskId, runDurationMillis
);
+
+ // The Indexer serves this task's log over HTTP straight from its own
process log file
+ // (see HttpRemoteTaskRunner#streamTaskLog), so the stream can become
present before the
+ // expected line has actually been written to that file. Poll until the
expected line shows
+ // up, rather than just until the stream is present, to avoid racing
against that lag.
+ final String logs = cluster.callApi().waitForResult(
+ () -> {
+ final Optional<InputStream> streamOptional = overlord.bindings()
+
.getInstance(TaskLogStreamer.class)
+
.streamTaskLog(taskId, 0);
+ if (!streamOptional.isPresent()) {
+ return "";
+ }
+ try (InputStream stream = streamOptional.get()) {
+ return IOUtils.toString(stream, StandardCharsets.UTF_8);
+ }
+ },
+ logContent -> logContent.contains(expectedLogLine)
+ ).go();
+
Assertions.assertFalse(logs.isEmpty());
Assertions.assertTrue(logs.contains(expectedLogLine), "Actual logs are: "
+ logs);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
