tkaymak opened a new pull request, #40932: URL: https://github.com/apache/beam/pull/40932
Part of #36841. When one unbounded source PCollection feeds two branches of a streaming query, for example a Flatten of a PCollection with itself, Spark does not deduplicate streaming relations and plans one scan per branch. Both scans open a reader for the same split in the same micro batch, so they shared one `BeamReaderCache` entry: the second task either took over the reader in flight or invalidated it. With bounded input the end of stream row was then emitted again on every batch and the query never went idle. The shared entry exists since #39971, the repeated end of stream row since #40766. The second scan now gets a detached reader, restored from the start mark of the batch, never cached and writing no marks or done markers. A sibling is recognised by the query run id, so a restarted query in the same JVM still restores from the durable mark. The detached reader inherits whether the end of stream row was already emitted. A failed mark write now also drops the cached reader. Sibling scans still read independently. A batch cut by `maxBatchDurationMillis` can differ per sibling on a live source, tracked in #40786. Bounded inputs that drain within the deadline are exact. `PAssertStreamingTest.testFlattenSameSource` flattens `Create` with itself and asserts every element twice. It hangs without the fix and passed 3 of 3 runs with it. Full `:runners:spark:4:test` and `:runners:spark:3:test` pass locally. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
