Skip to content

Race Condition Test for Iterator vs Staged Commits Deletion #5892

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Draft
TimothyW553 wants to merge 10 commits into
base: master
Choose a base branch
from
Draft

Race Condition Test for Iterator vs Staged Commits Deletion #5892

TimothyW553 wants to merge 10 commits into from

Conversation

@TimothyW553
Copy link
Collaborator

@TimothyW553 TimothyW553 commented Jan 20, 2026

Which Delta project/connector is this regarding?

  • Spark
  • Standalone
  • Flink
  • Kernel
  • Other (fill in here)

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

@TimothyW553
Copy link
Collaborator Author

TimothyW553 commented Jan 20, 2026 

[info] Test io.sparkuctest.UCDeltaStreamingTest#allTableTypesTestsFactory() #1:testStreamingWriteToManagedTable(EXTERNAL) started
[info] Test io.sparkuctest.UCDeltaStreamingTest#allTableTypesTestsFactory() #1:testStreamingWriteToManagedTable(MANAGED) started
[info] Test io.sparkuctest.UCDeltaStreamingTest#allTableTypesTestsFactory() #2:testStreamingReadFromTable(EXTERNAL) started
[info] Test io.sparkuctest.UCDeltaStreamingTest#allTableTypesTestsFactory() #2:testStreamingReadFromTable(MANAGED) started
[error] Test io.sparkuctest.UCDeltaStreamingTest.allTableTypesTestsFactory failed: java.lang.reflect.InvocationTargetException: null, took 1.506s
[error]     at java.lang.reflect.Method.invoke(Method.java:569)
[error]     at io.sparkuctest.UCDeltaTableIntegrationBaseTest.lambda$allTableTypesTestsFactory$1(UCDeltaTableIntegrationBaseTest.java:83)
[error]     at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
[error]     at java.util.stream.Streams$StreamBuilderImpl.forEachRemaining(Streams.java:411)
[error]     at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
[error]     at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276)
[error]     at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
[error]     at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
[error]     at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
[error]     at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
[error]     at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
[error]     at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
[error]     at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
[error]     at java.util.stream.ReferencePipeline.forEachOrdered(ReferencePipeline.java:601)
[error]     at java.util.Optional.ifPresent(Optional.java:178)
[error]     at java.util.ArrayList.forEach(ArrayList.java:1511)
[error]     at java.util.ArrayList.forEach(ArrayList.java:1511)
[error] Caused by: org.apache.spark.sql.streaming.StreamingQueryException: Encountered an error from the underlying engine implementation while trying to Error reading JSON file: file:/tmp/unity-catalog-test-3067490179900378433/ucroot/tables/62e41178-24ce-41ba-ad70-b7c1b724aa55/_delta_log/_staged_commits/00000000000000000003.164e8770-50d1-4fad-a1e3-d636cfd3bb21.json: null
[error] === Streaming Query ===
[error] Identifier: uc_streaming_read_managed_70eb3bf5924a417bb78207679d14981f [id = 7235e9c8-4338-48f5-b820-b50285b792a1, runId = 70e21138-5cd9-4dbb-a07a-29de84799383]
[error] Current Committed Offsets: {io.delta.spark.internal.v2.read.SparkMicroBatchStream@9316dbe: {"sourceVersion":1,"reservoirId":"84882ad2-f179-44d2-a2e5-690d9552d97e","reservoirVersion":4,"index":-1,"isStartingVersion":false}}
[error] Current Available Offsets: {io.delta.spark.internal.v2.read.SparkMicroBatchStream@9316dbe: {"sourceVersion":1,"reservoirId":"84882ad2-f179-44d2-a2e5-690d9552d97e","reservoirVersion":4,"index":-1,"isStartingVersion":false}}
[error] 
[error] Current State: TERMINATED
[error] Thread State: RUNNABLE
[error] 
[error] Logical Plan:
[error] ~WriteToMicroBatchDataSource MemorySink, 7235e9c8-4338-48f5-b820-b50285b792a1, [queryName=uc_streaming_read_managed_70eb3bf5924a417bb78207679d14981f, checkpointLocation=/tmp/spark-checkpoint-14837387079894795573], Append
[error] +- ~SubqueryAlias unity.default.streaming_read_test
[error]    +- ~StreamingDataSourceV2ScanRelation[id#2704L, value#2705] unity.default.streaming_read_test
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:372)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:226)
[error] Caused by: io.delta.kernel.exceptions.KernelEngineException: Encountered an error from the underlying engine implementation while trying to Error reading JSON file: file:/tmp/unity-catalog-test-3067490179900378433/ucroot/tables/62e41178-24ce-41ba-ad70-b7c1b724aa55/_delta_log/_staged_commits/00000000000000000003.164e8770-50d1-4fad-a1e3-d636cfd3bb21.json: null
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:117)
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:111)
[error]     at io.delta.kernel.utils.CloseableIterator$1.hasNext(CloseableIterator.java:107)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.combine(ActionsIterator.java:549)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.readCommitOrCompactionFile(ActionsIterator.java:516)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.getNextActionsIter(ActionsIterator.java:445)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.tryEnsureNextActionsIterIsReady(ActionsIterator.java:272)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.hasNext(ActionsIterator.java:214)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.loadProtocolAndMetadataInternal(ProtocolMetadataLogReplay.java:128)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.lambda$loadProtocolAndMetadata$0(ProtocolMetadataLogReplay.java:82)
[error]     at io.delta.kernel.internal.metrics.Timer.time(Timer.java:79)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.loadProtocolAndMetadata(ProtocolMetadataLogReplay.java:81)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.createSnapshot(SnapshotFactory.java:201)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.lambda$create$5(SnapshotFactory.java:165)
[error]     at io.delta.kernel.internal.metrics.Timer.time(Timer.java:79)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.create(SnapshotFactory.java:165)
[error]     at io.delta.kernel.internal.table.SnapshotBuilderImpl.build(SnapshotBuilderImpl.java:120)
[error]     at io.delta.kernel.internal.table.SnapshotBuilderImpl.build(SnapshotBuilderImpl.java:46)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.lambda$loadSnapshot$4(UCCatalogManagedClient.java:170)
[error]     at io.delta.kernel.internal.metrics.Timer.timeChecked(Timer.java:104)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.lambda$loadSnapshot$5(UCCatalogManagedClient.java:139)
[error]     at io.delta.kernel.internal.metrics.Timer.timeChecked(Timer.java:104)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.loadSnapshot(UCCatalogManagedClient.java:121)
[error]     at io.delta.spark.internal.v2.snapshot.unitycatalog.UCManagedTableSnapshotManager.loadLatestSnapshot(UCManagedTableSnapshotManager.java:69)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.filterDeltaLogs(SparkMicroBatchStream.java:678)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getFileChanges(SparkMicroBatchStream.java:619)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getFileChangesWithRateLimit(SparkMicroBatchStream.java:577)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getNextOffsetFromPreviousOffset(SparkMicroBatchStream.java:370)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.latestOffsetInternal(SparkMicroBatchStream.java:333)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.latestOffset(SparkMicroBatchStream.java:319)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:618)
[error]     at org.apache.spark.sql.execution.streaming.ProgressContext.reportTimeTaken(ProgressReporter.scala:186)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:617)
[error]     at scala.collection.immutable.List.map(List.scala:236)
[error]     at scala.collection.immutable.List.map(List.scala:79)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:606)
[error]     at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.scala:17)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:1011)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:602)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$executeOneBatch$2(MicroBatchExecution.scala:375)
[error]     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
[error]     at org.apache.spark.sql.execution.streaming.ProgressContext.reportTimeTaken(ProgressReporter.scala:186)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.executeOneBatch(MicroBatchExecution.scala:364)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1$adapted(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.TriggerExecutor.runOneBatch(TriggerExecutor.scala:39)
[error]     at org.apache.spark.sql.execution.streaming.TriggerExecutor.runOneBatch$(TriggerExecutor.scala:37)
[error]     at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.runOneBatch(TriggerExecutor.scala:70)
[error]     at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:82)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:337)
[error]     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
[error]     at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:804)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:311)
[error]     ... 1 more
[error] Caused by: java.nio.channels.ClosedByInterruptException
[error]     at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:199)
[error]     at sun.nio.ch.FileChannelImpl.endBlocking(FileChannelImpl.java:171)
[error]     at sun.nio.ch.FileChannelImpl.position(FileChannelImpl.java:373)
[error]     at org.apache.hadoop.fs.RawLocalFileSystem$LocalFSFileInputStream.seek(RawLocalFileSystem.java:183)
[error]     at org.apache.hadoop.fs.BufferedFSInputStream.seek(BufferedFSInputStream.java:105)
[error]     at org.apache.hadoop.fs.FSDataInputStream.seek(FSDataInputStream.java:73)
[error]     at org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker.readChunk(ChecksumFileSystem.java:281)
[error]     at org.apache.hadoop.fs.FSInputChecker.readChecksumChunk(FSInputChecker.java:305)
[error]     at org.apache.hadoop.fs.FSInputChecker.read1(FSInputChecker.java:257)
[error]     at org.apache.hadoop.fs.FSInputChecker.read(FSInputChecker.java:202)
[error]     at java.io.DataInputStream.read(DataInputStream.java:151)
[error]     at io.delta.kernel.defaults.engine.hadoopio.HadoopSeekableInputStream.read(HadoopSeekableInputStream.java:42)
[error]     at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:287)
[error]     at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:330)
[error]     at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:190)
[error]     at java.io.InputStreamReader.read(InputStreamReader.java:177)
[error]     at java.io.BufferedReader.fill(BufferedReader.java:162)
[error]     at java.io.BufferedReader.readLine(BufferedReader.java:329)
[error]     at java.io.BufferedReader.readLine(BufferedReader.java:396)
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:107)
[error]     ... 54 more
[info] Test run finished: 1 failed, 0 ignored, 4 total, 22.079s
[error] Failed: Total 4, Failed 1, Errors 0, Passed 3
[error] Failed tests:
[error] 	io.sparkuctest.UCDeltaStreamingTest
[error] (sparkUnityCatalog / Test / testOnly) sbt.TestsFailedException: Tests unsuccessful

@openinx
Copy link
Collaborator

openinx commented Jan 20, 2026

You can use this command to run against the remote uc server.

export UC_REMOTE=true
export UC_URI=$UC_URI
export UC_CATALOG_NAME=main
export UC_SCHEMA_NAME=demo_zh
export UC_BASE_TABLE_LOCATION=$S3_BASE_LOCATION

build/sbt "sparkUnityCatalog/testOnly io.sparkuctest.UCDeltaStreamingTest"

@TimothyW553
Copy link
Collaborator Author

TimothyW553 commented Jan 20, 2026 •
edited
Loading 

[info] Test io.sparkuctest.UCDeltaStreamingTest#allTableTypesTestsFactory() #1:testStreamingReadFromTable(MANAGED) started
[info] DEBUG[after-seed] table=unity.default.streaming_read_test location=file:///tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64 latest=1 published=/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/00000000000000000001.json publishedExists=true stagedDirExists=true
[info] DEBUG[after-seed] stagedFiles=[.00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json.crc, 00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json]
[info] DEBUG[after-insert-1] table=unity.default.streaming_read_test location=file:///tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64 latest=2 published=/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/00000000000000000002.json publishedExists=true stagedDirExists=true
[info] DEBUG[after-insert-1] stagedFiles=[.00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json.crc, .00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json.crc, 00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json, 00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json]
[info] DEBUG[after-process-1] table=unity.default.streaming_read_test location=file:///tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64 latest=2 published=/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/00000000000000000002.json publishedExists=true stagedDirExists=true
[info] DEBUG[after-process-1] stagedFiles=[.00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json.crc, .00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json.crc, 00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json, 00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json]
[info] DEBUG[after-insert-2] table=unity.default.streaming_read_test location=file:///tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64 latest=3 published=/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/00000000000000000003.json publishedExists=true stagedDirExists=true
[info] DEBUG[after-insert-2] stagedFiles=[.00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json.crc, .00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json.crc, .00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json.crc, 00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json, 00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json, 00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json]
[info] DEBUG[after-process-2] table=unity.default.streaming_read_test location=file:///tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64 latest=3 published=/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/00000000000000000003.json publishedExists=true stagedDirExists=true
[info] DEBUG[after-process-2] stagedFiles=[.00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json.crc, .00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json.crc, .00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json.crc, 00000000000000000001.25965b78-7d07-49f2-aa52-7d6cc7f4bd66.json, 00000000000000000002.011d326d-4b6e-4c74-a1ef-ae0a0fb926b8.json, 00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json]
[error] Test io.sparkuctest.UCDeltaStreamingTest.allTableTypesTestsFactory failed: java.lang.reflect.InvocationTargetException: null, took 7.845s
[error]     at java.lang.reflect.Method.invoke(Method.java:569)
[error]     at io.sparkuctest.UCDeltaTableIntegrationBaseTest.lambda$allTableTypesTestsFactory$1(UCDeltaTableIntegrationBaseTest.java:83)
[error]     at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
[error]     at java.util.stream.Streams$StreamBuilderImpl.forEachRemaining(Streams.java:411)
[error]     at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
[error]     at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276)
[error]     at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
[error]     at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
[error]     at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
[error]     at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
[error]     at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
[error]     at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
[error]     at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
[error]     at java.util.stream.ReferencePipeline.forEachOrdered(ReferencePipeline.java:601)
[error]     at java.util.Optional.ifPresent(Optional.java:178)
[error]     at java.util.ArrayList.forEach(ArrayList.java:1511)
[error]     at java.util.ArrayList.forEach(ArrayList.java:1511)
[error] Caused by: org.apache.spark.sql.streaming.StreamingQueryException: Encountered an error from the underlying engine implementation while trying to Error reading JSON file: file:/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/_staged_commits/00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json: null
[error] === Streaming Query ===
[error] Identifier: uc_streaming_read_managed_0fca983e66ea48e2b10366d6f9990e89 [id = 34a88f0e-6ce6-4f55-a299-36881d6b3f0e, runId = 4fa22af3-2f18-4d14-81af-1725946f26e9]
[error] Current Committed Offsets: {io.delta.spark.internal.v2.read.SparkMicroBatchStream@710e644f: {"sourceVersion":1,"reservoirId":"bbcf6cfd-95d3-4b23-9695-29263b1b448a","reservoirVersion":4,"index":-1,"isStartingVersion":false}}
[error] Current Available Offsets: {io.delta.spark.internal.v2.read.SparkMicroBatchStream@710e644f: {"sourceVersion":1,"reservoirId":"bbcf6cfd-95d3-4b23-9695-29263b1b448a","reservoirVersion":4,"index":-1,"isStartingVersion":false}}
[error] 
[error] Current State: TERMINATED
[error] Thread State: RUNNABLE
[error] 
[error] Logical Plan:
[error] ~WriteToMicroBatchDataSource MemorySink, 34a88f0e-6ce6-4f55-a299-36881d6b3f0e, [queryName=uc_streaming_read_managed_0fca983e66ea48e2b10366d6f9990e89, checkpointLocation=/tmp/spark-checkpoint-13834995782859400752], Append
[error] +- ~SubqueryAlias unity.default.streaming_read_test
[error]    +- ~StreamingDataSourceV2ScanRelation[id#175L, value#176] unity.default.streaming_read_test
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:372)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:226)
[error] Caused by: io.delta.kernel.exceptions.KernelEngineException: Encountered an error from the underlying engine implementation while trying to Error reading JSON file: file:/tmp/unity-catalog-test-430509415751217815/ucroot/tables/8081696c-1e77-4188-8564-82e7ec1add64/_delta_log/_staged_commits/00000000000000000003.752f7dd4-7fcb-41e3-9338-b70dbd01587e.json: null
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:117)
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:111)
[error]     at io.delta.kernel.utils.CloseableIterator$1.hasNext(CloseableIterator.java:107)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.combine(ActionsIterator.java:549)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.readCommitOrCompactionFile(ActionsIterator.java:516)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.getNextActionsIter(ActionsIterator.java:445)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.tryEnsureNextActionsIterIsReady(ActionsIterator.java:272)
[error]     at io.delta.kernel.internal.replay.ActionsIterator.hasNext(ActionsIterator.java:214)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.loadProtocolAndMetadataInternal(ProtocolMetadataLogReplay.java:128)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.lambda$loadProtocolAndMetadata$0(ProtocolMetadataLogReplay.java:82)
[error]     at io.delta.kernel.internal.metrics.Timer.time(Timer.java:79)
[error]     at io.delta.kernel.internal.replay.ProtocolMetadataLogReplay.loadProtocolAndMetadata(ProtocolMetadataLogReplay.java:81)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.createSnapshot(SnapshotFactory.java:201)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.lambda$create$5(SnapshotFactory.java:165)
[error]     at io.delta.kernel.internal.metrics.Timer.time(Timer.java:79)
[error]     at io.delta.kernel.internal.table.SnapshotFactory.create(SnapshotFactory.java:165)
[error]     at io.delta.kernel.internal.table.SnapshotBuilderImpl.build(SnapshotBuilderImpl.java:120)
[error]     at io.delta.kernel.internal.table.SnapshotBuilderImpl.build(SnapshotBuilderImpl.java:46)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.lambda$loadSnapshot$4(UCCatalogManagedClient.java:170)
[error]     at io.delta.kernel.internal.metrics.Timer.timeChecked(Timer.java:104)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.lambda$loadSnapshot$5(UCCatalogManagedClient.java:139)
[error]     at io.delta.kernel.internal.metrics.Timer.timeChecked(Timer.java:104)
[error]     at io.delta.kernel.unitycatalog.UCCatalogManagedClient.loadSnapshot(UCCatalogManagedClient.java:121)
[error]     at io.delta.spark.internal.v2.snapshot.unitycatalog.UCManagedTableSnapshotManager.loadLatestSnapshot(UCManagedTableSnapshotManager.java:69)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.filterDeltaLogs(SparkMicroBatchStream.java:678)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getFileChanges(SparkMicroBatchStream.java:619)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getFileChangesWithRateLimit(SparkMicroBatchStream.java:577)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.getNextOffsetFromPreviousOffset(SparkMicroBatchStream.java:370)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.latestOffsetInternal(SparkMicroBatchStream.java:333)
[error]     at io.delta.spark.internal.v2.read.SparkMicroBatchStream.latestOffset(SparkMicroBatchStream.java:319)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:618)
[error]     at org.apache.spark.sql.execution.streaming.ProgressContext.reportTimeTaken(ProgressReporter.scala:186)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:617)
[error]     at scala.collection.immutable.List.map(List.scala:236)
[error]     at scala.collection.immutable.List.map(List.scala:79)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:606)
[error]     at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.scala:17)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:1011)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:602)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$executeOneBatch$2(MicroBatchExecution.scala:375)
[error]     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
[error]     at org.apache.spark.sql.execution.streaming.ProgressContext.reportTimeTaken(ProgressReporter.scala:186)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.executeOneBatch(MicroBatchExecution.scala:364)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1$adapted(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.TriggerExecutor.runOneBatch(TriggerExecutor.scala:39)
[error]     at org.apache.spark.sql.execution.streaming.TriggerExecutor.runOneBatch$(TriggerExecutor.scala:37)
[error]     at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.runOneBatch(TriggerExecutor.scala:70)
[error]     at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:82)
[error]     at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:344)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:337)
[error]     at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
[error]     at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:804)
[error]     at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:311)
[error]     ... 1 more
[error] Caused by: java.nio.channels.ClosedByInterruptException
[error]     at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:199)
[error]     at sun.nio.ch.FileChannelImpl.endBlocking(FileChannelImpl.java:171)
[error]     at sun.nio.ch.FileChannelImpl.position(FileChannelImpl.java:373)
[error]     at org.apache.hadoop.fs.RawLocalFileSystem$LocalFSFileInputStream.seek(RawLocalFileSystem.java:183)
[error]     at org.apache.hadoop.fs.BufferedFSInputStream.seek(BufferedFSInputStream.java:105)
[error]     at org.apache.hadoop.fs.FSDataInputStream.seek(FSDataInputStream.java:73)
[error]     at org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker.readChunk(ChecksumFileSystem.java:281)
[error]     at org.apache.hadoop.fs.FSInputChecker.readChecksumChunk(FSInputChecker.java:305)
[error]     at org.apache.hadoop.fs.FSInputChecker.read1(FSInputChecker.java:257)
[error]     at org.apache.hadoop.fs.FSInputChecker.read(FSInputChecker.java:202)
[error]     at java.io.DataInputStream.read(DataInputStream.java:151)
[error]     at io.delta.kernel.defaults.engine.hadoopio.HadoopSeekableInputStream.read(HadoopSeekableInputStream.java:42)
[error]     at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:287)
[error]     at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:330)
[error]     at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:190)
[error]     at java.io.InputStreamReader.read(InputStreamReader.java:177)
[error]     at java.io.BufferedReader.fill(BufferedReader.java:162)
[error]     at java.io.BufferedReader.readLine(BufferedReader.java:329)
[error]     at java.io.BufferedReader.readLine(BufferedReader.java:396)
[error]     at io.delta.kernel.defaults.engine.DefaultJsonHandler$1.hasNext(DefaultJsonHandler.java:107)
[error]     ... 54 more
[info] Test run finished: 1 failed, 0 ignored, 1 total, 15.134s
[error] Failed: Total 1, Failed 1, Errors 0, Passed 0
[error] Failed tests:
[error] 	io.sparkuctest.UCDeltaStreamingTest
[error] (sparkUnityCatalog / Test / testOnly) sbt.TestsFailedException: Tests unsuccessful
[error] Total time: 45 s, completed Jan 20, 2026, 11:15:13 PM

@TimothyW553 TimothyW553 mentioned this pull request Jan 21, 2026
Merged
5 tasks
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@TimothyW553 @openinx