The current implementation of @RequiresTimeSortedInput uses BagState, which is read and sorted upon watermark update. This can be improved once SortedMapState is available.
Imported from Jira BEAM-9256. Original Jira may contain additional context.
Reported by: janl.
