(fluss) branch main updated: [spark] Fix spark async write exception propagation (#4332)

[yu...@apache.org]

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

luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 03e511a29 [spark] Fix spark async write exception propagation (#4332)
03e511a29 is described below

commit 03e511a2974c747ffd36612c1cef95ba6aaeaca8
Author: Yang Zhang <ya...@alibaba-inc.com>
AuthorDate: Wed Sep 16 10:09:29 2026 +0800

    [spark] Fix spark async write exception propagation (#4332)
---
 .../apache/fluss/spark/write/FlussDataWriter.scala |  2 +-
 .../org/apache/fluss/spark/SparkWriteTest.scala    | 73 +++++++++++++++++++++-
 2 files changed, 71 insertions(+), 4 deletions(-)

diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/write/FlussDataWriter.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/write/FlussDataWriter.scala
index 94fae6133..9fd6676e3 100644
--- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/write/FlussDataWriter.scala
+++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/write/FlussDataWriter.scala
@@ -55,7 +55,7 @@ abstract class FlussDataWriter[T](
   protected val flussRow = new SparkAsFlussRow(dataSchema)
 
   @volatile
-  private var asyncWriterException: Option[Throwable] = _
+  private var asyncWriterException: Option[Throwable] = None
 
   def writeRow(record: SparkAsFlussRow): CompletableFuture[T]
 
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkWriteTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkWriteTest.scala
index a2382c161..1198a5ada 100644
--- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkWriteTest.scala
+++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkWriteTest.scala
@@ -17,15 +17,22 @@
 
 package org.apache.fluss.spark
 
-import org.apache.fluss.metadata.{Schema, TableDescriptor}
+import org.apache.fluss.client.table.writer.{AppendResult, TableWriter}
+import org.apache.fluss.config.Configuration
+import org.apache.fluss.metadata.{Schema, TableDescriptor, TablePath}
 import org.apache.fluss.row.{BinaryString, GenericRow, InternalRow}
+import org.apache.fluss.spark.row.SparkAsFlussRow
 import org.apache.fluss.spark.util.TestUtils.{createGenericRow, FLUSS_ROWTYPE}
+import org.apache.fluss.spark.write.{FlussDataWriter, FlussWriterCommitMessage}
 import org.apache.fluss.types.DataTypes
 
-import org.assertj.core.api.Assertions.assertThat
+import org.apache.spark.sql.catalyst.{InternalRow => SparkInternalRow}
+import org.apache.spark.sql.types.StructType
+import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
 
+import java.io.IOException
 import java.sql.Timestamp
-import java.time.Duration
+import java.util.concurrent.CompletableFuture
 
 import scala.collection.JavaConverters._
 
@@ -151,6 +158,66 @@ class SparkWriteTest extends FlussSparkTestBase {
     assertThat(flussRows2.length).isEqualTo(4)
     assertThat(flussRows2).containsAll(expectRows2.toIterable.asJava)
   }
+
+  test("asynchronous failure is propagated on the next write") {
+    val result = new CompletableFuture[AppendResult]()
+    val writer = createWriter(result)
+    val failure = new IOException("Fluss append failed")
+
+    writer.write(SparkInternalRow.empty)
+    result.completeExceptionally(failure)
+
+    assertThatThrownBy(() => writer.write(SparkInternalRow.empty))
+      .isInstanceOf(classOf[IOException])
+      .hasCause(failure)
+  }
+
+  test("asynchronous failure during flush is propagated on commit") {
+    val result = new CompletableFuture[AppendResult]()
+    val failure = new IOException("Fluss append failed")
+    val writer = createWriter(
+      result,
+      () => {
+        result.completeExceptionally(failure)
+        ()
+      })
+
+    writer.write(SparkInternalRow.empty)
+
+    assertThatThrownBy(() => writer.commit())
+      .isInstanceOf(classOf[IOException])
+      .hasCause(failure)
+  }
+
+  test("successful asynchronous write can commit") {
+    val result = new CompletableFuture[AppendResult]()
+    val writer = createWriter(
+      result,
+      () => {
+        result.complete(new AppendResult())
+        ()
+      })
+
+    writer.write(SparkInternalRow.empty)
+
+    assertThat(writer.commit()).isEqualTo(FlussWriterCommitMessage())
+    assertThat(result.isDone).isTrue
+  }
+
+  private def createWriter(
+      result: CompletableFuture[AppendResult],
+      onFlush: () => Unit = () => ()): FlussDataWriter[AppendResult] = {
+    new FlussDataWriter[AppendResult](
+      TablePath.of("test", "log_table"),
+      new StructType(),
+      new Configuration()) {
+      override val writer: TableWriter = new TableWriter {
+        override def flush(): Unit = onFlush()
+      }
+
+      override def writeRow(record: SparkAsFlussRow): CompletableFuture[AppendResult] = result
+    }
+  }
 }
 
 object SparkWriteTest {