Commit 2da27cb
Implement Asynchronous wrapper for DoFn in Java SDK (apache#38609)
* Created an Asynchronous Wrapper for DoFn as well as JUnit tests for the Apache Beam Java SDK (apache#38529)
* Optimize State reconciliation loop and eliminate O(N^2) complexity. Removed O(N) global activeElements scan. Fixed logic bug where duplicate elements were incorrectly marked for rescheduling. Optimized lookups by converting finishedItems from a list to a HashSet.
* Added check for long overflow possibility when exponentially increasing sleep. Added two more tests to match Python SDK. Fixed formatting issues.
* Fix Timestamp propagation and add relevant test too. Spotless Apply fixes. Spot Bugs potential fixes.
* Resolve SpotBugs DMI_RANDOM_USED_ONLY_ONCE replacing new Random(seed) that preserves deterministic jitter behavior and avoids pressure on garbage collector (apache#38529)
* Improve AsyncDoFn robustness and fix critical warnings provided by gemini-code-assist
- Propagates asynchronous task exceptions as RuntimeExceptions to prevent silent data loss and enable runner-level retries.
- Implements a static refCounts registry to safely tear down the shared executor service only when the last cloned instance is destroyed.
- Validates timerFrequency in the constructor to prevent zero/negative values from entering infinite loops.
- Documents multi-threading requirements, multi-output limitations, and bundle lifecycle behaviors in a class-level comment.
* Fixed formatting issue
* Implement cross-key task cancellation in AsyncDoFn.
Stores the partition key inside InFlightElement and cancels/purges orphaned
futures inside commitFinishedItems. This prevents silent memory leaks on
bundle rollbacks (apache#38529)
* Passes original element's inputTimestamp to AccumulatingOutputReceiver to preserve event-time downstream. Initializes the ExecutorService thread pool only when useThreadPool is true. Refactors state filtering to use finishedElementIds instead of finishedItems, preventing duplicate processing (apache#38529)
* Changed outputs from List type to ConcurrentLinkedQueue to prevent lock contention (apache#38529)
* Reverted outputs back to list type. Takes original Bounded Window as parameter without Hardcoding GlobalWindow.Instance (apache#38529)
* Refactor AsyncDoFn to simplify output receiver and fix type mismatches involving inputTimestamp(apache#38529)
* Rename AsyncDoFn to PerKeyConcurrentDoFn to address naming feedback. Might also need to rename python sdk too.
* Final Formatting fixes. (apache#38529)
* Revert class, file, test naming back to AsyncDoFn. Extracted verbose ArgumentProvider code block into util functions. Refactored verbose logging into util functions. Replaced hardcoded numbers with static constants. Updated test helper methods with @VisibleForTesting (apache#38529)
* Rename class and files to AsyncWrapper and clean up old duplicates (apache#38529)
* Refactor verbose logging to a single logInfo utility function (apache#38529)1 parent cd76624 commit 2da27cb
2 files changed
Lines changed: 1657 additions & 0 deletions
File tree
- sdks/java/core/src
- main/java/org/apache/beam/sdk/transforms
- test/java/org/apache/beam/sdk/transforms
0 commit comments