This is an automated email from the ASF dual-hosted git repository.
uros-b 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 54ff9ed1dd66 [SPARK-57365][SS][TEST] Fix flaky KafkaRelationSuite
resolved-offset test by using distinct message timestamps
54ff9ed1dd66 is described below
commit 54ff9ed1dd66e885ccb4c781f71c1255428c5723
Author: Iván Morales <[email protected]>
AuthorDate: Sun Jun 21 00:52:19 2026 +0200
[SPARK-57365][SS][TEST] Fix flaky KafkaRelationSuite resolved-offset test
by using distinct message timestamps
### What changes were proposed in this pull request?
Make the test `"resolved start offset greater than end offset (without
latest)"` in `KafkaRelationSuite` deterministic. The two messages per partition
are now produced with explicit, increasing CreateTime timestamps (`base`, `base
+ 1`) via `RecordBuilder`, so the second message (offset 1) is
deterministically resolved by `offsetsForTimes`. The `eventually(60.seconds)`
wrapper is also removed, since the resolved-offset check fires synchronously on
the driver and retrying could not cha [...]
### Why are the changes needed?
The test is flaky. It produces two messages per partition with no explicit
timestamp and uses the second message's CreateTime as the ending-offset
timestamp, expecting `offsetsForTimes` to resolve to offset 1. When both
messages land in the same millisecond they share a timestamp, `offsetsForTimes`
resolves to offset 0, and the hardcoded `"resolved end offset 1"` assertion
fails with `"resolved end offset 0"`. The existing `eventually(60.seconds)`
does not help: produced timestamps ar [...]
### Does this PR introduce _any_ user-facing change?
No. Test-only change.
### How was this patch tested?
A/B, 50 runs x 4 suite variants (V1 / V2 / WithAdminV1 / WithAdminV2) = 200
test instances per arm:
- With the fix: 200/200 passed.
- Forcing the race (both messages produced with the same timestamp)
reproduces the failure deterministically: `"The resolved start offset 3 is
greater than the resolved end offset 0 ..."` vs the expected `"... resolved end
offset 1 ..."`.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude (Anthropic), used for analysis, code generation and
review assistance.
Generated-by: Claude Sonnet 4.6
Closes #56427 from kete1987/SPARK-57365-kafka-relation-flaky.
Authored-by: Iván Morales <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit b88af0a02a03da577fefa0cec4c218a4937b0971)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../spark/sql/kafka010/KafkaRelationSuite.scala | 33 +++++++++++++---------
1 file changed, 19 insertions(+), 14 deletions(-)
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRelationSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRelationSuite.scala
index 0d4e4f8fb79a..ab515e2a8e49 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRelationSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRelationSuite.scala
@@ -21,8 +21,6 @@ import java.nio.charset.StandardCharsets.UTF_8
import java.util.Locale
import java.util.concurrent.atomic.AtomicInteger
-import scala.concurrent.duration.DurationInt
-
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.TopicPartition
@@ -675,12 +673,21 @@ abstract class KafkaRelationSuiteBase extends
SharedSparkSession with KafkaTest
test("resolved start offset greater than end offset (without latest)") {
val topic = newTopic()
testUtils.createTopic(topic, partitions = 3)
- val timestamp1 =
- testUtils.sendMessages(topic, Seq("0", "0").toArray,
Some(0))(1)._2.timestamp()
- val timestamp2 =
- testUtils.sendMessages(topic, Seq("0", "0").toArray,
Some(1))(1)._2.timestamp()
- val timestamp3 =
- testUtils.sendMessages(topic, Seq("0", "0").toArray,
Some(2))(1)._2.timestamp()
+ // Two messages per partition with explicit, increasing timestamps so the
second message
+ // (offset 1) is deterministically resolved by offsetsForTimes. Without
explicit timestamps
+ // both messages can share a CreateTime millisecond and resolve to offset
0, making the test
+ // flaky (produced timestamps are fixed at produce time, so retrying
always resolves the same
+ // wrong offset). Returns base + 1 directly; broker echoes the producer
timestamp unchanged
+ // under the default CreateTime semantics.
+ def sendTwoWithDistinctTs(part: Int): Long = {
+ val base = System.currentTimeMillis()
+ testUtils.sendMessages(Seq(
+ new RecordBuilder(topic, "0").partition(part).timestamp(base).build(),
+ new RecordBuilder(topic, "0").partition(part).timestamp(base +
1).build()
+ ))
+ base + 1
+ }
+ val Seq(timestamp1, timestamp2, timestamp3) = (0 to
2).map(sendTwoWithDistinctTs)
val df = spark.read
.format("kafka")
.option("kafka.bootstrap.servers", testUtils.brokerAddress)
@@ -699,13 +706,11 @@ abstract class KafkaRelationSuiteBase extends
SharedSparkSession with KafkaTest
)
.load()
- eventually(timeout(60.seconds)) {
- val e = intercept[IllegalStateException] {
- df.collect()
- }
- assert(e.getMessage.contains(
- "The resolved start offset 3 is greater than the resolved end offset 1
for"))
+ val e = intercept[IllegalStateException] {
+ df.collect()
}
+ assert(e.getMessage.contains(
+ "The resolved start offset 3 is greater than the resolved end offset 1
for"))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
