This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 64f2739492c7 [SPARK-57947][SS][KAFKA][TEST] Deflake
KafkaSourceStressForDontFailOnDataLossSuite
64f2739492c7 is described below
commit 64f2739492c7cdea3b496b605c77e6e5d75ef4b2
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Mon Jul 6 17:23:57 2026 +0900
[SPARK-57947][SS][KAFKA][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. This PR widens
the offset-fetch retry budget the test configures:
```
- .option("fetchOffset.retryIntervalMs", "3000")
+ .option("fetchOffset.numRetries", "10")
+ .option("fetchOffset.retryIntervalMs", "500")
```
(from the default `numRetries=3` and the test's previous
`retryIntervalMs=3000`, i.e. ~3s of retrying, to `10 × 500ms` = up to ~5s).
### Why are the changes needed?
The test is flaky — it fails roughly **50%** of runs on the
`build_branch42_maven_java21` lane and occasionally on master, always with:
```
org.apache.spark.sql.streaming.StreamingQueryException: [STREAM_FAILED] ...
terminated with exception:
org.apache.kafka.common.errors.UnknownTopicOrPartitionException: Failed to
fetch metadata for
partition failOnDataLoss-<n>-0 because metadata for topic
`failOnDataLoss-<n>` could not be found
```
Root cause: an offset fetch races one of the test's own concurrent topic
deletions. `KafkaOffsetReader{Admin,Consumer}.withRetries` retries transient
`NonFatal` failures, but only `fetchOffset.numRetries` times (default 3).
During a burst of the test's deliberate topic churn, 3 retries can be exhausted
before the topic's metadata becomes available again, so the transient
`UnknownTopicOrPartitionException` is rethrown and fails the query — even
though `failOnDataLoss=false`, which is e [...]
Giving the reader more retries at a shorter interval lets the test ride out
its self-induced churn (a deleted topic is typically recreated within a second)
without exhausting the budget and without materially slowing the query. This
does not change the production `failOnDataLoss` semantics; it only tunes the
test's own reader options.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
Because the failure is intermittent (~50%), the `kafka` Maven lane was run
repeatedly on a fork via GitHub Actions.
**Before (apache/spark, FAILING):**
- `Build / Maven (branch-4.2, JDK 21)`:
https://github.com/apache/spark/actions/runs/28749011147 —
`KafkaSourceStressForDontFailOnDataLossSuite: stress test for
failOnDataLoss=false *** FAILED ***` with the
`UnknownTopicOrPartitionException` above.
**After (HyukjinKwon fork, PASSING — Maven JDK21, matrix narrowed to the
`kafka-0-10*` modules; 3/3 green, `stress test for failOnDataLoss=false` passes
and the `kafka-0-10-sql` suite reports `Tests: succeeded 561, failed 0`):**
- https://github.com/HyukjinKwon/spark/actions/runs/28762085085
- https://github.com/HyukjinKwon/spark/actions/runs/28762086851
- https://github.com/HyukjinKwon/spark/actions/runs/28762088779
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57022 from HyukjinKwon/ci-fix/agent3-kafka-stress-retries.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
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 b46d2d2d0624..6a65cde7a7fc 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
@@ -229,7 +229,16 @@ class KafkaSourceStressForDontFailOnDataLossSuite extends
StreamTest with KafkaM
.option("subscribePattern", "failOnDataLoss.*")
.option("startingOffsets", "earliest")
.option("failOnDataLoss", "false")
- .option("fetchOffset.retryIntervalMs", "3000")
+ // 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")
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]
