SparkStateInternals backs MapState and SetState with one coded cell per state. remove() writes the shrunk collection back even when it is now empty, see SparkStateInternals.java lines 432 to 436 (map) and 537 to 541 (set). isEmpty() only checks whether the cell exists (lines 501 and 560), so it returns false after the last element is removed. InMemoryStateInternals returns true in that case.
The shared StateInternalsTest does not catch this, because it checks isEmpty() only after clear().
Side effect: SparkGroupAlsoByWindowViaWindowSet keeps a key's state as long as getState() is not empty (lines 265 and 307). A key whose map or set was emptied by removals is therefore kept instead of dropped.
Fix: when the collection becomes empty, call clear() instead of writing it back. Add one test to StateInternalsTest for isEmpty after removing the last map and set element.
Line numbers refer to master 00e322188bf. Found while working on #36841.
SparkStateInternalsbacksMapStateandSetStatewith one coded cell per state.remove()writes the shrunk collection back even when it is now empty, seeSparkStateInternals.javalines 432 to 436 (map) and 537 to 541 (set).isEmpty()only checks whether the cell exists (lines 501 and 560), so it returns false after the last element is removed.InMemoryStateInternalsreturns true in that case.The shared
StateInternalsTestdoes not catch this, because it checksisEmpty()only afterclear().Side effect:
SparkGroupAlsoByWindowViaWindowSetkeeps a key's state as long asgetState()is not empty (lines 265 and 307). A key whose map or set was emptied by removals is therefore kept instead of dropped.Fix: when the collection becomes empty, call
clear()instead of writing it back. Add one test toStateInternalsTestfor isEmpty after removing the last map and set element.Line numbers refer to master
00e322188bf. Found while working on #36841.