Gaurav598 commented on issue #39631: URL: https://github.com/apache/beam/issues/39631#issuecomment-5312708339
Hi, I’ve completed a deeper investigation of the SDF execution path and wanted to share the proposed direction before starting the implementation. One important finding is that bounded SDFs also require residual handling: @BoundedPerElement DoFns can return ProcessContinuation.resume(), which produces DelayedBundleApplication residuals. After adding the standard SplittableParDoExpander.createSizedReplacement() translation step diagnostically, the Kafka Streams runner reaches bundle execution but fails at bundle completion with UnsupportedOperationException: The ActiveBundle does not have a registered bundle checkpoint handler. The implementation direction I’m considering is: expand SDF ParDos before fusion using SplittableParDoExpander.createSizedReplacement(); reuse StateAndTimerBundleCheckpointHandler; back residual state/timers with the runner’s existing Kafka Streams state/timer stores; add processing-time timer draining via a Kafka Streams wall-clock punctuator; replay residuals through the executable stage’s main input receiver; then enable the bounded SDF ValidatesRunner tests. I also verified that the shared checkpoint handler uses ValueState, so no new MapState support should be required. Since #39770 is now merging the Kafka Streams runner into master, should I base this work on that merge branch / wait for it to land, or continue against feat/18479-kafka-streams-runner-skeleton for now? I’m happy to proceed with #39631 if this approach matches the intended direction. -- 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]
