feat(java): worker — dispatch bridge, handlers, events#300
Merged
Conversation
JavaDispatcher implements WorkerDispatcher with a registry-in-Rust completion bridge: jobs submit to Java via onJob and resolve through NativeWorker.completeJob/failJob/cancelJob. Wires scheduler, result drain, and worker lifecycle.
Queue.worker() builder registers typed handlers and runs them via the WorkerBridge/WorkerControl SPI; outcomes fan out through an Emitter. Renames the registered-worker view to WorkerInfo.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
✅ Files skipped from review due to trivial changes (1)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughWalkthroughAdds Java and Rust support for running Taskito workers through JNI: new worker and event contracts, a Java worker builder and dispatch bridge, native queue/control entrypoints, Rust-side dispatch and worker lifecycle code, plus updated docs and an end-to-end worker test. ChangesWorker runtime bridge
Sequence Diagram(s)sequenceDiagram
participant Queue
participant WorkerBuilder
participant JniQueueBackend
participant NativeQueue
participant JniWorkerControl
Queue->>WorkerBuilder: worker()
WorkerBuilder->>JniQueueBackend: startWorker(...)
JniQueueBackend->>NativeQueue: runWorker(handle, bridge, optionsJson)
NativeQueue-->>JniQueueBackend: native worker handle
JniQueueBackend-->>WorkerBuilder: JniWorkerControl
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 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: 8
🧹 Nitpick comments (1)
sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java (1)
25-49: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider adding coverage for non-success outcomes.
This validates only the happy path. The bridge also handles failures (
failJobon handler exception / missing handler), cancellation, and timeout, which map toEventName.RETRY/DEAD/CANCELLED. A follow-up test (e.g. a handler that throws, or a missing-handler task) would exercise the failure dispatch path andOutcomeEventemission.Want me to draft the failure-path test case?
🤖 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/WorkerTest.java` around lines 25 - 49, Add a test in WorkerTest that covers a non-success path instead of only the happy path; the current runsTaskToCompletion test only verifies EventName.SUCCESS. Create a case using the existing Worker/Queue flow where the handler throws or no handler is registered for the task, then assert the failure outcome is dispatched (for example EventName.RETRY, DEAD, or CANCELLED) and that the corresponding OutcomeEvent is emitted. Use the existing queue.worker().handle(...).on(...) pattern so the new test is easy to locate and consistent with the current coverage.
🤖 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/dispatcher.rs`:
- Around line 90-92: The async task in dispatcher::dispatch currently calls
crossbeam_channel::Sender::send directly inside tokio::spawn, which can block
Tokio worker threads when the bounded result channel is full. Move the result
handoff in the run_one completion path off the async worker by using a
non-blocking approach or delegating the blocking send to a separate blocking
context, and keep the tokio task focused on awaiting run_one and returning
promptly.
- Around line 164-178: The WorkerBridge.onJob call in dispatcher.rs is letting a
Java exception escape before it is cleared, which can leave the attached thread
in a bad state. Update the exception handling around env.call_method in the
onJob path so the Error::JavaException case is handled first, clear the pending
exception with env.exception_clear(), and then return the Rust error; keep the
existing exception_check/exception_clear flow for any remaining post-call
validation.
In `@crates/taskito-java/src/worker.rs`:
- Around line 346-368: `Java_org_byteveda_taskito_internal_NativeWorker_close`
is reclaiming the native `WorkerHandle` too early, while Java-side handlers can
still call JNI entrypoints like `completeJob` and `failJob` after
`Worker.close()`. Make the close path graceful and idempotent by stopping new
scheduling first, waiting for in-flight Java callbacks to quiesce and lifecycle
unregister to finish, and only then calling `drop_handle::<WorkerHandle>`;
alternatively keep a closed/ref-counted state alive so late completions are
ignored safely instead of borrowing a freed pointer.
- Around line 157-173: The call_on_outcome path is returning too early from
JNIEnv::call_method, which can leave a Java exception pending on the reused
JNIEnv. Update the call to onOutcome in WorkerBridge so that call_method errors
are handled explicitly before using ?/return, then check and clear any pending
exception with exception_check/exception_clear and map both failure paths into a
single Err. Keep the fix localized around call_on_outcome and the
env.call_method invocation.
- Around line 131-170: The drain loop in `drain_results` and the JNI helper
`call_on_outcome` are creating local JNI references for every result without
releasing them, which can exhaust the local-reference table under load. Update
the per-outcome path so each iteration uses a local frame or `auto_local` for
the `JNIEnv` string/object references created in `call_on_outcome`, and ensure
they are dropped before the next `scheduler.handle_result`/`onOutcome` call.
In `@sdks/java/README.md`:
- Around line 36-43: The Java README worker example is self-deadlocking because
Worker.awaitShutdown() is called inside a try-with-resources block, while the
implicit close() only happens after the block exits. Update the example around
Taskito.builder(), queue.worker(), and Worker.awaitShutdown() to demonstrate an
external shutdown trigger or a shutdown hook that explicitly closes the worker
so the block can exit cleanly before resource cleanup.
In `@sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java`:
- Around line 7-35: Make JniWorkerControl close-safe by preventing
NativeWorker.close(handle) from being called more than once. The issue is that
the handle field stays usable after close(), so repeated AutoCloseable paths can
double-release the native resource; update JniWorkerControl to guard the native
handle and make close() idempotent, while keeping completeJob, failJob,
cancelJob, and stop from using a closed handle.
In `@sdks/java/src/main/java/org/byteveda/taskito/Worker.java`:
- Around line 51-56: The Worker.close() sequence is closing the native control
handle too early while executor work may still be running. Update Worker.close()
to stop accepting new work, drain or wait for outstanding executor tasks started
via WorkerDispatchBridge.runJob, and only then call control.close() before
counting down shutdown. Use the existing control, executor, and shutdown symbols
in Worker to preserve a safe close order.
---
Nitpick comments:
In `@sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java`:
- Around line 25-49: Add a test in WorkerTest that covers a non-success path
instead of only the happy path; the current runsTaskToCompletion test only
verifies EventName.SUCCESS. Create a case using the existing Worker/Queue flow
where the handler throws or no handler is registered for the task, then assert
the failure outcome is dispatched (for example EventName.RETRY, DEAD, or
CANCELLED) and that the corresponding OutcomeEvent is emitted. Use the existing
queue.worker().handle(...).on(...) pattern so the new test is easy to locate and
consistent with the current coverage.
🪄 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: 3baa4f7e-0323-4498-943d-30d57415674f
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (26)
crates/taskito-java/Cargo.tomlcrates/taskito-java/src/convert.rscrates/taskito-java/src/dispatcher.rscrates/taskito-java/src/jvm.rscrates/taskito-java/src/lib.rscrates/taskito-java/src/worker.rssdks/java/README.mdsdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.javasdks/java/src/main/java/org/byteveda/taskito/Queue.javasdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.javasdks/java/src/main/java/org/byteveda/taskito/TaskFunction.javasdks/java/src/main/java/org/byteveda/taskito/Worker.javasdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.javasdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.javasdks/java/src/main/java/org/byteveda/taskito/events/Emitter.javasdks/java/src/main/java/org/byteveda/taskito/events/EventName.javasdks/java/src/main/java/org/byteveda/taskito/events/OutcomeEvent.javasdks/java/src/main/java/org/byteveda/taskito/events/package-info.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.javasdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.javasdks/java/src/main/java/org/byteveda/taskito/spi/WorkerBridge.javasdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.javasdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java
💤 Files with no reviewable changes (1)
- crates/taskito-java/src/jvm.rs
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.
Second of the Java SDK split series (follows #299). Adds the worker runtime so tasks execute end-to-end.
What's here
dispatcher.rs+worker.rs. A registry-in-Rust async bridge: the scheduler submits a job to Java via a cheaponJob(token, …)callback and awaits aoneshot; Java runs the handler and completes viacompleteJob/failJob/cancelJob. No tokio worker thread blocks. The worker owns its own multi-thread runtime, drives the coreScheduler+ dispatcher, drains outcomes back to Java (onOutcome), and runs the heartbeat/lifecycle.WorkerBridge(Rust→Java:onJob/onOutcome) andWorkerControl(Java→Rust:completeJob/failJob/cancelJob/stop/close);internal/{NativeWorker, JniWorkerControl}.Worker+Worker.Builder(.handle(Task, fn),.concurrency,.on(EventName, listener),.start()),TaskFunction<T,R>,RegisteredTask,WorkerDispatchBridge(runs the handler, serializes the result, completes; awaits aCompletableFuture<WorkerControl>to avoid the onJob-before-control race).Queue.worker()returns the builder.events/{EventName, OutcomeEvent, Emitter}.Notes
Worker→WorkerInfo(this PR'sWorkeris the runtime);listWorkers()returnsList<WorkerInfo>.Verification
cargo fmt + clippy (taskito-java) clean; Java compiles under
--release 11(main +WorkerTest). The dedicated Java CI job arrives later in the series; the Rust workspace is covered by existing CI.Summary by CodeRabbit
New Features
WorkerControl/WorkerBridge.WorkerInfo,EventName,OutcomeEvent,Emitter).Bug Fixes
Documentation