You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "wuchong (via GitHub)" <gi...@apache.org> on 2026/04/18 08:42:38 UTC
[I] [Test] Unstable test SparkStreamingTest read: log table [fluss]
wuchong opened a new issue, #3129:
URL: https://github.com/apache/fluss/issues/3129
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Fluss version
0.9.0 (latest release)
### Please describe the bug 🐞
https://github.com/apache/fluss/actions/runs/24599047572/job/71934470469?pr=2976
```
SparkStreamingTest:
- write: write to log table
- write: write to primary key table
- read: log table *** FAILED ***
Error while stopping stream:
query.exception() is not empty after clean stop: org.apache.spark.sql.streaming.StreamingQueryException: Failed to update metadata
=== Streaming Query ===
Identifier: [id = 4a10ba53-3612-4f36-8c2b-edc03172db04, runId = 888a7078-87a2-45b1-94e8-9f4b8d63b915]
Current Committed Offsets: {org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311: {"version":1,"table_id":2,"bucket_offsets":[3]}}
Current Available Offsets: {org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311: {"version":1,"table_id":2,"bucket_offsets":[3]}}
Current State: TERMINATED
Thread State: RUNNABLE
Logical Plan:
WriteToMicroBatchDataSource MemorySink, 4a10ba53-3612-4f36-8c2b-edc03172db04, Append
+- SubqueryAlias fluss_catalog.fluss.t
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311
== Progress ==
AssertOnQuery(<condition>, )
CheckLastBatch:
=> StopStream
StartStream(ProcessingTimeTrigger(0),org.apache.spark.util.SystemClock@65520982,Map(),null)
AddFlussData(t,StructType(StructField(id,IntegerType,true),StructField(data,StringType,true)),List([4,data4], [5,data5]))
AssertOnQuery(<condition>, )
CheckAnswer: [4,data4],[5,data5]
== Stream ==
Output Mode: Append
Stream state: {org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311: {"version":1,"table_id":2,"bucket_offsets":[3]}}
Thread state: dead
== Sink ==
0:
== Plan ==
== Parsed Logical Plan ==
WriteToMicroBatchDataSource MemorySink, 4a10ba53-3612-4f36-8c2b-edc03172db04, Append, 0
+- SubqueryAlias fluss_catalog.fluss.t
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Analyzed Logical Plan ==
WriteToMicroBatchDataSource MemorySink, 4a10ba53-3612-4f36-8c2b-edc03172db04, Append, 0
+- SubqueryAlias fluss_catalog.fluss.t
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Optimized Logical Plan ==
WriteToDataSourceV2 MicroBatchWrite[epoch: 0, writer: org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@69150bfc]
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Physical Plan ==
WriteToDataSourceV2 MicroBatchWrite[epoch: 0, writer: org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@69150bfc], org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$16044/0x0000000804c11440@60737ee1
+- *(1) Project [id#2277, data#2278]
+- MicroBatchScan[id#2277, data#2278] class org.apache.fluss.spark.read.FlussAppendScan
org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
org.scalatest.funsuite.AnyFunSuite.newAssertionFailedException(AnyFunSuite.scala:1562)
org.scalatest.Assertions.fail(Assertions.scala:933)
org.scalatest.Assertions.fail$(Assertions.scala:929)
org.scalatest.funsuite.AnyFunSuite.fail(AnyFunSuite.scala:1562)
org.apache.spark.sql.streaming.StreamTest.failTest$1(StreamTest.scala:462)
org.apache.spark.sql.streaming.StreamTest.verify$1(StreamTest.scala:433)
org.apache.spark.sql.streaming.StreamTest.$anonfun$testStream$26(StreamTest.scala:606)
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
== Progress ==
AssertOnQuery(<condition>, )
CheckLastBatch:
=> StopStream
StartStream(ProcessingTimeTrigger(0),org.apache.spark.util.SystemClock@65520982,Map(),null)
AddFlussData(t,StructType(StructField(id,IntegerType,true),StructField(data,StringType,true)),List([4,data4], [5,data5]))
AssertOnQuery(<condition>, )
CheckAnswer: [4,data4],[5,data5]
== Stream ==
Output Mode: Append
Stream state: {org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311: {"version":1,"table_id":2,"bucket_offsets":[3]}}
Thread state: dead
== Sink ==
0:
== Plan ==
== Parsed Logical Plan ==
WriteToMicroBatchDataSource MemorySink, 4a10ba53-3612-4f36-8c2b-edc03172db04, Append, 0
+- SubqueryAlias fluss_catalog.fluss.t
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Analyzed Logical Plan ==
WriteToMicroBatchDataSource MemorySink, 4a10ba53-3612-4f36-8c2b-edc03172db04, Append, 0
+- SubqueryAlias fluss_catalog.fluss.t
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Optimized Logical Plan ==
WriteToDataSourceV2 MicroBatchWrite[epoch: 0, writer: org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@69150bfc]
+- StreamingDataSourceV2Relation [id#2277, data#2278], fluss_catalog.fluss.t, FlussAppendScan(fluss.t,TableInfo{tablePath=fluss.t, tableId=2, schemaId=1, schema=Schema{columns=[id INT, data STRING], primaryKey=null, autoIncrementColumnNames=[], highestFieldId=1}, physicalPrimaryKeys=[], bucketKeys=[], partitionKeys=[], numBuckets=1, properties={table.replication.factor=1}, customProperties={owner=fluss}, remoteDataDir=null, comment='null', createdTime=1776497158880, modifiedTime=1776497158880},None,org.apache.spark.sql.util.CaseInsensitiveStringMap@4d5141e1,{bootstrap.servers=127.0.0.1:44085}), org.apache.fluss.spark.read.FlussAppendMicroBatchStream@249ff311, {"version":1,"table_id":2,"bucket_offsets":[3]}, {"version":1,"table_id":2,"bucket_offsets":[3]}
== Physical Plan ==
WriteToDataSourceV2 MicroBatchWrite[epoch: 0, writer: org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@69150bfc], org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$16044/0x0000000804c11440@60737ee1
+- *(1) Project [id#2277, data#2278]
+- MicroBatchScan[id#2277, data#2278] class org.apache.fluss.spark.read.FlussAppendScan (StreamTest.scala:462)
- read: log partition table
- read: primary key table
- read: primary key partition table
- ```
### Solution
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Test] Unstable test SparkStreamingTest read: log table [fluss]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on issue #3129:
URL: https://github.com/apache/fluss/issues/3129#issuecomment-4277722173
> cc [@Yohahaha](https://github.com/Yohahaha)
I will try fix it this week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Test] Unstable test SparkStreamingTest read: log table [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on issue #3129:
URL: https://github.com/apache/fluss/issues/3129#issuecomment-4273260597
cc @Yohahaha
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Test] Unstable test SparkStreamingTest read: log table [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia closed issue #3129: [Test] Unstable test SparkStreamingTest read: log table
URL: https://github.com/apache/fluss/issues/3129
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Test] Unstable test SparkStreamingTest read: log table [fluss]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on issue #3129:
URL: https://github.com/apache/fluss/issues/3129#issuecomment-4281201388
## Root Cause
**Race condition** between Spark's `StreamExecution.stop()` and Fluss metadata fetch.
### Timeline (from logs)
```
1013594 [ScalaTest-main] StopStream
1013595 [ScalaTest-main] Asked to cancel job group
1013598 [stream exec] ERROR Query terminated
Caused by: InterruptedException
at CompletableFuture.get
at MetadataUpdater.updateMetadata
```
### Sequence Diagram (Before Fix)
```
Test Main Thread Stream Execution Thread
│ │
│ testStream(...) │ MicroBatchExecution.runActivatedStream
│ │ │ while (isActive) {
│ │ ProcessAllAvailable() │ constructNextBatch()
│ │ ───────────────────────────────┼──→ latestOffset()
│ │ │ → fetchLatestOffsets()
│ │ │ → FlussAdmin.listOffsets()
│ │ │ → CompletableFuture.get() ← blocking
│ │ │
│ StopStream │
│ │ │
│ query.stop() │
│ state.set(TERMINATED) │
│ cancelJobGroup(runId) │
│ interruptThread() ────────────────┼──→ Thread.interrupt()
│ queryExecutionThread.join() │
│ │ CompletableFuture.get()
│ │ → throws InterruptedException
│ │ ↓
│ │ MetadataUpdater.updateMetadata()
│ │ catch block:
│ │ not RetriableException
│ │ not PartitionNotExistException
│ │ → throw FlussRuntimeException
│ │ ↓
│ │ Spark catch block:
│ │ isInterruptedByStop()?
│ │ FlussRuntimeException → NOT recognized
│ │ → query.exception() = this exception
│ │
│ stopSources() │
│ → FlussMicroBatchStream.stop() │ (thread already terminated)
│ │
│ check query.exception() → NOT EMPTY! → TEST FAILS
```
### Sequence Diagram (After Fix)
```
Test Main Thread Stream Execution Thread
│ │
│ testStream(...) │ MicroBatchExecution.runActivatedStream
│ │ │ while (isActive) {
│ │ ProcessAllAvailable() │ constructNextBatch()
│ │ ───────────────────────────────┼──→ latestOffset()
│ │ │ → fetchLatestOffsets()
│ │ │ → FlussAdmin.listOffsets()
│ │ │ → CompletableFuture.get() ← blocking
│ │ │
│ StopStream │
│ │ │
│ query.stop() │
│ state.set(TERMINATED) │
│ cancelJobGroup(runId) │
│ interruptThread() ────────────────┼──→ Thread.interrupt()
│ queryExecutionThread.join() │
│ │ CompletableFuture.get()
│ │ → throws InterruptedException
│ │ ↓
│ │ latestOffset() catch block:
│ │ isInterruptionException(e)? → YES
│ │ → logWarning("interrupted during shutdown")
│ │ → return start offset ← clean exit
│ │ ↓
│ │ Spark catch block:
│ │ no exception thrown
│ │ → query.exception() = None
│ │
│ stopSources() │
│ → FlussMicroBatchStream.stop() │ (thread already terminated)
│ │
│ check query.exception() → EMPTY! → TEST PASSES
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org