(spark) branch branch-4.2 updated: [SPARK-58017][SS][TEST] Deflake KafkaSourceStressForDontFailOnDataLossSuite

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

HyukjinKwon pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.2 by this push:
     new 35277aa0f4f9 [SPARK-58017][SS][TEST] Deflake KafkaSourceStressForDontFailOnDataLossSuite
35277aa0f4f9 is described below

commit 35277aa0f4f970b0b475441b2bc8e08e561ffa31
Author: Hyukjin Kwon <gu...@apache.org>
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 <gu...@apache.org>
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
    (cherry picked from commit 5557aa198d69c051271943c3f86eb8565e33b526)
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
 .../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org