feat(java): workflow fan-out / fan-in#306
Merged
Merged
Conversation
|
Warning Review limit reached
More reviews will be available in 37 minutes and 15 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (7)
📝 WalkthroughWalkthroughJava and Rust workflow bindings now carry fan-out/fan-in metadata, defer node execution, expose workflow plan and node-lookup APIs, and add runtime handling for expansion, aggregation, skip cascades, and terminal finalization. The README and a new end-to-end test cover the fan-out/fan-in flow. ChangesFan-out and fan-in workflow orchestration
Sequence Diagram(s)sequenceDiagram
participant Worker
participant WorkflowTracker
participant JniQueueBackend
participant NativeWorkflows
Worker->>WorkflowTracker: onSuccess / onDead OutcomeEvent
WorkflowTracker->>JniQueueBackend: workflowNodeForJobJson(jobId)
JniQueueBackend->>NativeWorkflows: workflowNodeForJob(handle, jobId)
NativeWorkflows-->>JniQueueBackend: {runId, nodeName}
JniQueueBackend-->>WorkflowTracker: node reference JSON
WorkflowTracker->>JniQueueBackend: getWorkflowPlanJson(runId)
JniQueueBackend->>NativeWorkflows: getWorkflowPlan(handle, runId)
NativeWorkflows-->>WorkflowTracker: plan JSON
alt fan-out parent completes
WorkflowTracker->>JniQueueBackend: expandFanOut(...)
JniQueueBackend->>NativeWorkflows: expand_fan_out(...)
else fan-out child completes
WorkflowTracker->>JniQueueBackend: checkFanOutCompletionJson(...)
JniQueueBackend->>NativeWorkflows: checkFanOutCompletion(...)
WorkflowTracker->>JniQueueBackend: createDeferredJob(...)
JniQueueBackend->>NativeWorkflows: createDeferredJob(...)
end
WorkflowTracker->>JniQueueBackend: cascadeSkipPending(...) / finalizeRunIfTerminal(...)
JniQueueBackend->>NativeWorkflows: cascadeSkipPending(...) / finalizeRunIfTerminal(...)
Estimated code review effort🎯 5 (Critical) | ⏱️ ~90+ minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
🧹 Nitpick comments (1)
sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java (1)
30-34: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake the fan-in assertion order-sensitive.
This only verifies the final sum, so any permutation of
[1,4,9,16]still passes. An order-sensitive fan-in check here would catch regressions in how child results are collected.Also applies to: 46-66
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java` around lines 30 - 34, The fan-in test in WorkflowFanOutTest only checks the aggregated sum, so it can miss regressions where child results are collected in the wrong order. Update the assertions around the fan-in path for the fanOut/fanIn workflow (including the related cases in the same test class) to verify the ordered list of square results before summing, using the Workflow.named("fanpipe") setup and the fanIn("sum", sum, "all", "square") output. Keep the sum check, but add an order-sensitive assertion on the collected child results so permutations fail.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/taskito-java/src/workflows/mod.rs`:
- Around line 721-729: The fan-in input order is currently taken from the
storage-return order in the workflow node lookup, which can make child_job_ids
nondeterministic. In the workflow fan-out handling around
get_workflow_nodes_by_prefix, sort the children by the numeric item index
encoded in the parent[index] node name before collecting job ids, so the fan-in
list matches the producer order consistently.
- Around line 207-208: The same-DAG reuse check in the workflow definition path
is only comparing dag_data, so stale step metadata can be reused even when
fan_out/fan_in changed. Update the existing-definition logic in mod.rs around
the workflow plan/definition comparison to also validate stored step_metadata
against the current step_meta, using the relevant StepMetadata fields such as
fan_out and fan_in. If the metadata differs, reject reuse with the existing
“bump the version” style error instead of returning the old definition.
- Around line 671-693: The child job enqueue flow in create_workflow_nodes_batch
currently makes each child visible before the corresponding workflow node
exists, which allows a worker to process it too early. Update the workflow
creation path around the child enqueue loop and wf.create_workflow_nodes_batch
so job insertion and node creation are atomic, or ensure child jobs stay
non-runnable until the node batch is successfully committed. Use the existing
symbols child_names, child_payloads, queue.storage.enqueue, and
create_workflow_nodes_batch to keep the enqueue-and-track steps synchronized.
- Around line 661-664: The zero-item fan-out path in `checkFanOutCompletion`/the
parent completion flow is inconsistent: empty children currently return no
completion signal, which can leave downstream fan-in waiting forever. Update the
logic in `checkFanOutCompletion` and the zero-count handling around
`wf.set_workflow_node_fan_out_count` / `wf.set_workflow_node_completed` so that
a recorded fan-out count of 0 is treated as a successful completion with an
empty `childJobIds` list, instead of returning null.
In `@sdks/java/README.md`:
- Line 110: Clarify the README guidance around trackWorkflows() so it states
that every Worker processing workflow jobs must opt in, not just one instance.
Update the wording near trackWorkflows() to explain that it only registers local
SUCCESS/DEAD listeners on the specific Worker, so fan-out expansion and
aggregation will not advance unless each worker in a multi-worker setup enables
it. Use the trackWorkflows() symbol and Worker concept to make the requirement
explicit.
In `@sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java`:
- Around line 128-157: The new workflow-tracker hooks added to QueueBackend are
currently abstract, which breaks existing custom backends. Update QueueBackend
so the new methods like getWorkflowPlanJson, workflowNodeForJobJson,
expandFanOut, checkFanOutCompletionJson, createDeferredJob, cascadeSkipPending,
and finalizeRunIfTerminal are provided as default methods that preserve source
compatibility, ideally by throwing UnsupportedOperationException as the other
optional SPI capabilities do, or move them into a separate capability interface.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java`:
- Line 15: The run plan is losing the workflow’s resolved queue and later falls
back to a hard-coded default, which can enqueue deferred jobs on the wrong
queue. Update the PlanNode/RunPlan flow so the resolved queue from submit time
is preserved per node or carried through the run-level plan, and then use that
stored value in the WorkflowTracker path instead of defaulting in the
fan-out/fan-in job creation logic.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java`:
- Around line 36-44: The RunPlan successor lookup currently returns only the
first matching deferred successor, which leaves additional fan-out successors
unprocessed. Update the matching helper in RunPlan (the successorMatching logic)
to return all PlanNode matches instead of a single node, then adjust the tracker
call sites that consume it to iterate over the full collection and route every
matching deferred successor.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java`:
- Around line 95-104: The Builder methods fanOut and fanIn on Step currently
accept arbitrary strings, so typos or conflicting values can silently produce
the wrong runtime behavior. Add validation in these setters to only allow the
supported strategies for each mode and reject invalid values, and also prevent a
Step from being configured as both fan-out and fan-in at the same time. Use the
Step.Builder fanOut and fanIn methods to locate the change and keep the runtime
branching aligned with the validated strategy values.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java`:
- Around line 48-57: Update Workflow.fanOut and Workflow.fanIn so deferred nodes
must be declared with exactly one after predecessor. Add validation in the
fanOut/fanIn methods (or the shared Step/Builder path they call) to reject zero
or multiple after values, and make sure the error clearly mentions the offending
workflow node name and strategy so misuse is caught before build/runtime.
In `@sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java`:
- Around line 78-80: The missing-result path in WorkflowTracker is being treated
as a normal return, but fetchResult() can return null on timeout or
interruption, so callers must not continue silently. Update the handling in
WorkflowTracker methods such as expandFanOutIfAny() and handleFanOutChild() so a
null from fetchResult() is treated as a retriable/fatal workflow state instead
of dropping expansion or fanning in a null value. Make the affected call sites
propagate or surface the failure consistently, and ensure the fan-out/fan-in
logic only proceeds once a real result is available.
---
Nitpick comments:
In `@sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java`:
- Around line 30-34: The fan-in test in WorkflowFanOutTest only checks the
aggregated sum, so it can miss regressions where child results are collected in
the wrong order. Update the assertions around the fan-in path for the
fanOut/fanIn workflow (including the related cases in the same test class) to
verify the ordered list of square results before summing, using the
Workflow.named("fanpipe") setup and the fanIn("sum", sum, "all", "square")
output. Keep the sum check, but add an order-sensitive assertion on the
collected child results so permutations fail.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 273479f5-e9a7-45ef-8c84-85cb8071a431
📒 Files selected for processing (13)
crates/taskito-java/src/workflows/mod.rssdks/java/README.mdsdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/worker/Worker.javasdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.javasdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.javasdks/java/src/main/java/org/byteveda/taskito/workflows/Step.javasdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.javasdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.javasdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java
…cers, missing result)
This was referenced Jun 26, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Seventh of the Java SDK split series (follows #305). Adds fan-out / fan-in to the workflow engine.
What's here
workflows.rs:expandFanOut,checkFanOutCompletion,createDeferredJob,cascadeSkipPending,finalizeRunIfTerminal,workflowNodeForJob,getWorkflowPlan.Workflow.fanOut(name, task, var, after…)maps a step over a producer's result list (one child job per item);Workflow.fanIn(name, task, var, after…)gathers the children's results into a list.WorkflowTracker: a lazy per-run plan cache (getWorkflowPlan, so submit and tracker aren't coupled); the deferred-node set is computed at submit, and the tracker expands fan-out at runtime, checks completion, creates deferred children, and finalizes the run.serialize(item), fan-in payload =serialize(List).Verification
Full
./gradlew build --no-daemonpasses: cargo--features postgres,redis,workflows+ compile + Spotless + Checkstyle + tests (incl.WorkflowFanOutTest: seed → [1,2,3,4] → square each → sum = 30), native bundled.Summary by CodeRabbit
New Features
Bug Fixes
Documentation