This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 88f62b6b6f43 [SPARK-58017][SS][TEST] Deflake
KafkaSourceStressForDontFailOnDataLossSuite
88f62b6b6f43 is described below
commit 88f62b6b6f4305937c699c262a2e4c291a77b623
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Wed Jul 8 08:56:42 2026 +0900
[SPARK-58017][SS][TEST] Deflake KafkaSourceStressForDontFailOnDataLossSuite
### What changes were proposed in this pull request?
Test-only change to deflake `KafkaSourceStressForDontFailOnDataLossSuite`
(the `stress test for failOnDataLoss=false` test).
The test rapidly creates, deletes and recreates topics (with
`kafka.metadata.max.age.ms=1`) while a streaming query with
`failOnDataLoss=false` runs, and asserts the query never fails. An offset fetch
can race one of the test's own concurrent topic deletions and hit
`UnknownTopicOrPartitionException`. That is transient and expected here
(`failOnDataLoss=false`): the offset reader retries until the metadata refresh
drops the deleted topic from the pattern subscription. But a bounded re [...]
Widen the reader's retry budget so it comfortably outlasts the test's own
20s stress loop:
```
- .option("fetchOffset.numRetries", "10")
- .option("fetchOffset.retryIntervalMs", "500")
+ .option("fetchOffset.numRetries", "60")
+ .option("fetchOffset.retryIntervalMs", "1000")
```
When the loop ends, `query.stop()` interrupts any in-flight retry cleanly
(the retry loop guards on `Thread.currentThread().isInterrupted` and
`Thread.sleep` throws `InterruptedException`), so the larger budget does not
slow teardown.
### Why are the changes needed?
The test is flaky on the `build_maven` lane (observed on branch-4.1),
failing with `UnknownTopicOrPartitionException` even though
`failOnDataLoss=false`.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
Ran `KafkaSourceStressForDontFailOnDataLossSuite` repeatedly on GitHub
Actions on a fork.
**CI results**
- Failing (before): https://github.com/apache/spark/actions/runs/28871397866
- Passing (after, 5/5 green):
https://github.com/HyukjinKwon/spark/actions/runs/28905834386
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57095 from HyukjinKwon/ci-fix/kafka-stress-deflake.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 5557aa198d69c051271943c3f86eb8565e33b526)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../sql/kafka010/KafkaDontFailOnDataLossSuite.scala | 16 +++++++++-------
1 file changed, 9 insertions(+), 7 deletions(-)
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
index 6a65cde7a7fc..2532c19092c4 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
@@ -232,13 +232,15 @@ class KafkaSourceStressForDontFailOnDataLossSuite extends
StreamTest with KafkaM
// This test deliberately creates, deletes and recreates topics as fast
as it can while a
// query is running. With Kafka metadata refreshed every 1ms, an offset
fetch can race a
// concurrent topic deletion and hit UnknownTopicOrPartitionException.
That is transient and
- // expected here, so the offset reader retries; but the default retry
budget (3 attempts x
- // 1s) can be exhausted during a burst of churn, letting the exception
surface and fail the
- // query even though failOnDataLoss=false. Give the reader more retries
at a short interval so
- // the test's own churn (a deleted topic is typically recreated within a
second) is ridden out
- // without exhausting the budget and without materially slowing the
query.
- .option("fetchOffset.numRetries", "10")
- .option("fetchOffset.retryIntervalMs", "500")
+ // expected here (failOnDataLoss=false): the offset reader just needs to
keep retrying until
+ // the metadata refresh drops the deleted topic from the pattern
subscription, after which the
+ // fetch succeeds. A bounded retry budget can still be exhausted during
a burst of churn under
+ // CI load (a previous 10 x 500ms = 5s budget still flaked), letting the
exception surface and
+ // fail the query. Give the reader a retry budget that comfortably
outlasts the test's own
+ // stress loop (20s below) so it rides out any churn window instead of
giving up; when the
+ // loop ends, query.stop() interrupts any in-flight retry cleanly.
+ .option("fetchOffset.numRetries", "60")
+ .option("fetchOffset.retryIntervalMs", "1000")
val kafka = reader.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
