This is an automated email from the ASF dual-hosted git repository.
amogh-jahagirdar pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 33cd2fd05f Spark: fix flaky testReadingStreamFromFutureTimestamp by
switching to Awaitility.await (#17405)
33cd2fd05f is described below
commit 33cd2fd05f8f0f711f6c215dc17e7147e92f4137
Author: Hongyue/Steve Zhang <[email protected]>
AuthorDate: Wed Jul 29 06:12:34 2026 -0700
Spark: fix flaky testReadingStreamFromFutureTimestamp by switching to
Awaitility.await (#17405)
---
.../spark/source/TestStructuredStreamingRead3.java | 16 ++++++++++------
.../spark/source/TestStructuredStreamingRead3.java | 16 ++++++++++------
.../spark/source/TestStructuredStreamingRead3.java | 16 ++++++++++------
3 files changed, 30 insertions(+), 18 deletions(-)
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 4efb883b5d..a36db04005 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -24,6 +24,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -75,6 +76,7 @@ import org.apache.spark.sql.streaming.DataStreamWriter;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.Trigger;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -602,7 +604,7 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
}
@TestTemplate
- public void testReadingStreamFromFutureTimetsamp() throws Exception {
+ public void testReadingStreamFromFutureTimestamp() throws Exception {
long futureTimestamp = System.currentTimeMillis() + 10000;
StreamingQuery query =
@@ -627,12 +629,14 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
waitUntilAfter(futureTimestamp);
- // Data appended after the timestamp should appear
+ // With async planning the new snapshot is discovered by a background
thread, so poll until
+ // the appended data becomes visible.
appendData(data);
- // Allow async background thread to refresh, else test sometimes fails
- Thread.sleep(50);
- actual = rowsAvailable(query);
- assertThat(actual).containsExactlyInAnyOrderElementsOf(data);
+ Awaitility.await("data appended after fromTimestamp becomes visible")
+ .atMost(Duration.ofSeconds(10))
+ .pollInterval(Duration.ofMillis(10))
+ .untilAsserted(
+ () ->
assertThat(rowsAvailable(query)).containsExactlyInAnyOrderElementsOf(data));
}
@TestTemplate
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 06189b3042..2cf728830b 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -24,6 +24,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -75,6 +76,7 @@ import org.apache.spark.sql.streaming.DataStreamWriter;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.Trigger;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -609,7 +611,7 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
}
@TestTemplate
- public void testReadingStreamFromFutureTimetsamp() throws Exception {
+ public void testReadingStreamFromFutureTimestamp() throws Exception {
long futureTimestamp = System.currentTimeMillis() + 10000;
StreamingQuery query =
@@ -634,12 +636,14 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
waitUntilAfter(futureTimestamp);
- // Data appended after the timestamp should appear
+ // With async planning the new snapshot is discovered by a background
thread, so poll until
+ // the appended data becomes visible.
appendData(data);
- // Allow async background thread to refresh, else test sometimes fails
- Thread.sleep(50);
- actual = rowsAvailable(query);
- assertThat(actual).containsExactlyInAnyOrderElementsOf(data);
+ Awaitility.await("data appended after fromTimestamp becomes visible")
+ .atMost(Duration.ofSeconds(10))
+ .pollInterval(Duration.ofMillis(10))
+ .untilAsserted(
+ () ->
assertThat(rowsAvailable(query)).containsExactlyInAnyOrderElementsOf(data));
}
@TestTemplate
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 89947f73ea..8003497bd1 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -24,6 +24,7 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -76,6 +77,7 @@ import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.Trigger;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -610,7 +612,7 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
}
@TestTemplate
- public void testReadingStreamFromFutureTimetsamp() throws Exception {
+ public void testReadingStreamFromFutureTimestamp() throws Exception {
long futureTimestamp = System.currentTimeMillis() + 10000;
StreamingQuery query =
@@ -635,12 +637,14 @@ public final class TestStructuredStreamingRead3 extends
CatalogTestBase {
waitUntilAfter(futureTimestamp);
- // Data appended after the timestamp should appear
+ // With async planning the new snapshot is discovered by a background
thread, so poll until
+ // the appended data becomes visible.
appendData(data);
- // Allow async background thread to refresh, else test sometimes fails
- Thread.sleep(50);
- actual = rowsAvailable(query);
- assertThat(actual).containsExactlyInAnyOrderElementsOf(data);
+ Awaitility.await("data appended after fromTimestamp becomes visible")
+ .atMost(Duration.ofSeconds(10))
+ .pollInterval(Duration.ofMillis(10))
+ .untilAsserted(
+ () ->
assertThat(rowsAvailable(query)).containsExactlyInAnyOrderElementsOf(data));
}
@TestTemplate
