You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/07/06 08:24:13 UTC
(spark) branch master updated: [SPARK-57947][SS][KAFKA][TEST] Deflake KafkaSourceStressForDontFailOnDataLossSuite
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 <gu...@apache.org>
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 <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org