Spark 3.2: Commit consumed offsets to the checkpoint location

[SreeramGarlapati]

This is Port of bug fix from our internal fork. Without this - our table maintenance expired the initial snapshot that was checkpoint'ed by the tables - which intern rendered our streaming readers to unrecoverable state.

cc: @rajarshisarkar, @daksha121, @rdblue, @aokolnychyi, @RussellSpitzer

[RussellSpitzer]

there should be a test with this patch which illustrates the issue being fixed if possible

[SreeramGarlapati]

@RussellSpitzer - thanks a lot for your review.

The challenge with unit testing this - is that - sparkSession is holding on to the state of the stream - internally - which makes it non-reproducible with 1 sparkSession. We saw this issue when our streams in EMR cluster had to be bought down and restarted on another EMR cluster & were able to verify this fix using the same approach.
So, all-in-all - I might be needing a unit test which spans across 2 sparkSessions & try this; the way tests are written is needing a bit of refactor to make it use 2 spark sessions & even after that I am unsure - if this will reproduce this (given the process sharing). Is there any precedence in the codebase for this pattern that you could kindly point me to.

I have put the fix before even I have that unit test before the public - for visibility - so ppl. can port it as needed. This bug is essentially a time bomb in the code (which blasts off the moment a stream moves across clusters) whoever is using readStream from iceberg table & have first snapshot expired; which is almost everyone who uses this!

Let me try that 2 spark sessions creation unittest and get back.

putting the problem being explained above aside - the code itself has coverage in this unittest - testResumingStreamReadFromCheckpoint - using which I was able to debug thru & fix the code - where I need to use file.createOrOverwrite() instead of the existing file.create().

cc: @rajarshisarkar

[rdblue]

@SreeramGarlapati can you please update the description with a clear statement of the bug and how this fixes it?

[rdblue]

I agree with @RussellSpitzer. This should have a test to verify the behavior.

[singhpk234]

Thanks @SreeramGarlapati !!!

[question] I have a doubt, can multiple streaming writing to same check-point location, result in nondeterministic state.

Let's say initial snapshot id was : snapshot(1)
Table state eventually (snapshot(1) -> snapshot(2) -> snapshot(3) -> snapshot(4))
Now stream 1 started in cluster 1 and read / commited till snapshot(2),
Now at the same time stream 2 started in cluster 2, it started from snapshot(2), before it could commit snapshot(3) , stream 1 committed snapshot(3) & snapshot(4) now when stream2 tried to commit snapshot(3) it will be overwriting the state of stream 1 or vice versa, starting state of new stream let's say stream3 in cluster 3 nondeterministic, as one stream is running ahead of other (stream 1 / stream 2).

Also earlier if two stream would have started in the same time (with no prev checkpoint / offset file) one would have failed since we did a file.create() now they can co-exist. Your thoughts ?

[SreeramGarlapati]

@singhpk234 - multiple spark streaming clusters cannot run based on the same checkpoint location.

[ajantha-bhat]

Closing this as Spark 3.2 is no longer supported in the latest version.

If issue exist on other spark version, please handle there.