java: MeshJob Stream subscription — subscribeEvents (Java)

[dhyansraj]

Summary

• Adds Stream subscription mode for MeshJob (Java vertical of MeshJob: event injection into running jobs — recv_event + send_event + Stream subscription + trace propagation #1032 follow-up): MeshJobs.subscribeEvents(jobId, options) -> EventSubscription returning a blocking Closeable iterator over Map<String, Object> events.
• Completes the polyglot trilogy started by MeshJob: Stream subscription — subscribe_events (Python) #1047 (Python subscribe_events) and ts: MeshJob Stream subscription — subscribeEvents (TypeScript) #1049 (TS subscribeEvents). Multiple subscribers can observe the same job's events independently via per-call cursors.
• Four layers: FFI mesh_job_proxy_list_events returning a {events, next_after} JSON envelope; JNR binding in MeshCore.java; Java SDK EventSubscription + SubscribeOptions + MeshJobs.subscribeEvents overloads reusing the existing LinkedHashMap-based LRU proxyCache from java: MeshJob event injection parity — recvEvent/sendEvent/postEvent #1045; integration test tc27_java extending the long-task fixtures with run_until_done (provider) and commission_subscribe_observer (consumer) capabilities.
• Java-specific value-add: EventSubscription implements Closeable. Try-with-resources gives Java something Python's asyncio.CancelledError provides naturally and TS can't replicate without AbortController plumbing — clean iterator shutdown that stops issuing new FFI long-polls.

Review Notes

Independent review caught 1 BLOCKER and 5 substantive WARNINGs (all addressed in the amended commit):

1. JobProxy lock serialized concurrent subscribers (BLOCKER) — every JobProxy method was wrapped in a single synchronized block. Two EventSubscription instances sharing one cached JobProxy would serialize their 30-second long-polls. Replaced with ReentrantReadWriteLock: close() takes the write lock; listEvents / sendEvent / status / await / cancel take the read lock and can run concurrently. The Rust core's JobProxy is Sync (Arc + String, no internal mutex) so concurrent FFI calls are safe — empirically validated by the fact that Python/TS verticals have always worked without any client-side lock.
2. closed field needed volatile — without it, cross-thread close() calls from arbitrary threads may not be observed by the iterator's while (!closed) loop. Textbook JMM hole. Now volatile.
3. SubscribeOptions.Builder lacked after >= 0 validation — fail-fast belongs at the builder. Now throws IllegalArgumentException for negative values.
4. Integration fixture leaked EventSubscription on poster exception — Thread.sleep / postEvent calls could throw without closing the subscription. Now wrapped in try-with-resources.
5. Fixture silently swallowed subscriber RuntimeException — JobNotFoundException / malformed payload errors produced empty observed_events with no diagnostic. Now logged via slf4j.
6. close() doesn't interrupt an in-flight FFI long-poll — the original JavaDoc overstated the close() contract as "the Java analog of Python's asyncio.CancelledError propagation". Rewrote honestly: close() flips a flag that stops future long-polls but does NOT interrupt the in-flight one (true interruption would require plumbing cancellation through FFI — out of scope). Documented the practical guidance: short longPoll for rapid shutdown.

Other findings (4 INFOs + 2 cosmetic WARNINGs) skipped as not load-bearing.

Closes #1050

Test plan

• Java unit tests: 80/80 pass (76 baseline + 4 new — 2 for SubscribeOptions builder validation, 2 for ReentrantReadWriteLock semantics — including a behavioral test that asserts two threads can enter the read lock simultaneously via getReadLockCount() == 2)
• Rust FFI tests: 23/23 pass (unchanged from initial implementation)
• src-tests 12/12 pass (image rebuilt with the new FFI symbol)
• uc23_meshjob_java integration suite: 27/27 pass (26 baseline + 1 new tc27_java; tc26 retry-flake unrelated, passed clean isolated)
• parse_ffi_timeout_secs reused at FFI boundary (NaN/Inf/negative-sentinel)
• LRU JobProxy cache reused from java: MeshJob event injection parity — recvEvent/sendEvent/postEvent #1045 (no new cache)
• Server-side types filter forwarded end-to-end (no client-side post-filter)
• next_after watermark advances cursor on empty pages (parity with MeshJob: Stream subscription — subscribe_events (Python) #1047 Python and ts: MeshJob Stream subscription — subscribeEvents (TypeScript) #1049 TS)
• Boolean-seq and missing-seq rejection (parity with sibling verticals)

Summary by CodeRabbit

• New Features

  • Added event subscription API to long-poll and stream job events with optional event-type filtering.
  • Introduced subscribeEvents() method with configurable options for cursor positioning and timeout behavior.
  • New iterator-based event stream interface for consuming events in applications.

• Bug Fixes & Improvements

  • Enhanced thread-safety for concurrent job operations.

[coderabbitai]

Warning

Rate limit exceeded

@dhyansraj has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 37 minutes and 7 seconds before requesting another review.

You’ve run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 48ad8b4a-7c99-40a6-8dd5-956b075de570

📥 Commits

Reviewing files that changed from the base of the PR and between 2747cae and 4e6ab36.

📒 Files selected for processing (11)

• src/runtime/core/include/mcp_mesh_core.h
• src/runtime/core/src/jobs_ffi.rs
• src/runtime/java/mcp-mesh-core/src/main/java/io/mcpmesh/core/MeshCore.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/EventSubscription.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/JobProxy.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/MeshJobs.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/SubscribeOptions.java
• src/runtime/java/mcp-mesh-sdk/src/test/java/io/mcpmesh/MeshJobsSubscribeEventsTest.java
• tests/integration/suites/uc23_meshjob_java/fixtures/long-task-consumer-java/src/main/java/com/example/longtaskconsumer/LongTaskConsumerApplication.java
• tests/integration/suites/uc23_meshjob_java/fixtures/long-task-provider-java/src/main/java/com/example/longtaskprovider/LongTaskProviderApplication.java
• tests/integration/suites/uc23_meshjob_java/tc27_subscribe_events_streams_observer_pattern_java/test.yaml

📝 Walkthrough

Walkthrough

The PR adds cursor-based job event subscription across C/Rust/Java layers, refactors JobProxy to use ReentrantReadWriteLock for concurrent FFI safety, and provides EventSubscription as a long-polling iterator with optional type filtering and internal event buffering.

Changes

Job Event Subscription Feature

Layer / File(s)	Summary
FFI Contract & Implementation src/runtime/core/include/mcp_mesh_core.h, src/runtime/core/src/jobs_ffi.rs, src/runtime/java/mcp-mesh-core/src/main/java/io/mcpmesh/core/MeshCore.java	New C header function mesh_job_proxy_list_events with cursor pagination (after), optional JSON type filter, long-poll timeout, and JSON envelope response (events + next_after watermark). Rust FFI implementation parses args, calls JobProxy::list_events, and serializes the result; null-handle guard test ensures consistent error behavior.
JobProxy Concurrency Refactor src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/JobProxy.java	Replaces synchronized(lock) + AtomicBoolean with ReentrantReadWriteLock and volatile boolean closed; all FFI methods (status, await, cancel, sendEvent, listEvents) acquire read lock, while close() acquires write lock to prevent handle freeing during in-flight calls; ensureOpen() validates under lock contract.
EventSubscription & SubscribeOptions APIs src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/EventSubscription.java, src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/SubscribeOptions.java	EventSubscription implements Iterator<Map<String, Object>> and Closeable for long-lived event streams; internally buffers parsed events from proxy.listEvents() calls and advances cursor using both per-event seq and registry-provided next_after watermark. SubscribeOptions encapsulates immutable subscription config (types filter, after cursor, longPoll duration) with defensive copying and builder validation.
MeshJobs Convenience API src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/MeshJobs.java	Two overloads for subscribeEvents(jobId) and subscribeEvents(jobId, options) that resolve registry URL, fetch/reuse cached JobProxy, and return a new EventSubscription bound to that proxy and options.
Unit Test Suite src/runtime/java/mcp-mesh-sdk/src/test/java/io/mcpmesh/MeshJobsSubscribeEventsTest.java	Reflective tests verifying API signatures, defaults, builder round-trip and immutability, input validation (null/empty jobId), close/iterator contract (close halts long-polling, is idempotent), LRU cache reuse, after non-negativity, and concurrency (two threads holding read lock concurrently via ReentrantReadWriteLock reflection).
Integration Test & Fixture Tools tests/integration/.../tc27_subscribe_events_streams_observer_pattern_java/test.yaml, tests/integration/.../long-task-consumer-java/src/.../LongTaskConsumerApplication.java, tests/integration/.../long-task-provider-java/src/.../LongTaskProviderApplication.java	End-to-end observer pattern: consumer tool (commissionSubscribeObserver) submits run_until_done job, spawns daemon subscriber thread using MeshJobs.subscribeEvents to collect work events until final marker, posts three work events, and returns response with job id, posted seqs, subscriber status, observed event list, and job result. Provider tool (runUntilDone) receives events in a loop until terminal event (payload final: true) or loop exhaustion, accumulating {seq, payload} entries. YAML test validates non-error response, subscriber completion without timeout, independent observer event counting (3 observed), producer processing (3 events), and payload round-trip.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

• dhyansraj/mcp-mesh#1046: Adds the same core JobProxy::list_events surface (Rust core + language bindings) used by this PR's event subscription API.
• dhyansraj/mcp-mesh#1050: Implements the same feature end-to-end (C-ABI FFI function, JNR binding, Java SDK support for subscribeEvents/EventSubscription).
• dhyansraj/mcp-mesh#1048: Implements TypeScript subscribeEvents that relies on the same JobProxy::list_events core behavior exposed in this PR.

Possibly related PRs

• dhyansraj/mcp-mesh#1047: Implements the core JobProxy::list_events backend and Python bindings that this PR's C FFI wraps and Java consumes.
• dhyansraj/mcp-mesh#1045: Extends jobs_ffi.rs with event-related FFI endpoints and shares the same parse_ffi_timeout_secs timeout parsing path.
• dhyansraj/mcp-mesh#1049: Implements job event stream pagination in TypeScript/N-API by wiring JobProxy::list_events into JavaScript, paralleling this PR's Java integration.

Poem

🐰 A cursor hops through job events, long-polling with care—
Read locks dance while writes await their fair share.
EventSubscription buffers the stream, close stops the flow,
From C through Rust to Java's observer show.
Test suites verify the observer pattern's might! 🎉

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely captures the primary change: adding Java SDK support for MeshJob event stream subscription via subscribeEvents API.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feature/1050-meshjob-subscribe-events-java

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🤖 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
`@src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/EventSubscription.java`:
- Around line 132-137: In EventSubscription (check around the code using
seqRaw/seqNum and next_after), reject fractional numeric values before advancing
the cursor: explicitly verify that seqRaw and next_after are integer-valued
(e.g., compare seqRaw.doubleValue() == seqRaw.longValue() or test for integral
subclasses) and throw a MeshException with a clear message if they are
fractional or not numeric; only then set long seq = seqNum.longValue() and
update cursor. Ensure the same check is applied to the next_after handling block
as pointed out (the second occurrence around lines 150-153).

In
`@src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/SubscribeOptions.java`:
- Around line 109-111: In SubscribeOptions.Builder.longPoll, validate the
argument by rejecting null and non-positive durations immediately: in the
Builder.longPoll(Duration longPoll) method (class SubscribeOptions.Builder) call
Objects.requireNonNull(longPoll, "...") and then check that longPoll.toMillis()
> 0 (or longPoll.isZero()/isNegative()) and throw an IllegalArgumentException
with a clear message if invalid; return this unchanged on success. This causes
fast-fail behavior for invalid longPoll values instead of deferring the error to
runtime.

🪄 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: ca5a9d44-f94e-465d-b6eb-81553a8a7aba

📥 Commits

Reviewing files that changed from the base of the PR and between 31817b0 and 2747cae.

📒 Files selected for processing (11)

• src/runtime/core/include/mcp_mesh_core.h
• src/runtime/core/src/jobs_ffi.rs
• src/runtime/java/mcp-mesh-core/src/main/java/io/mcpmesh/core/MeshCore.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/EventSubscription.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/JobProxy.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/MeshJobs.java
• src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/SubscribeOptions.java
• src/runtime/java/mcp-mesh-sdk/src/test/java/io/mcpmesh/MeshJobsSubscribeEventsTest.java
• tests/integration/suites/uc23_meshjob_java/fixtures/long-task-consumer-java/src/main/java/com/example/longtaskconsumer/LongTaskConsumerApplication.java
• tests/integration/suites/uc23_meshjob_java/fixtures/long-task-provider-java/src/main/java/com/example/longtaskprovider/LongTaskProviderApplication.java
• tests/integration/suites/uc23_meshjob_java/tc27_subscribe_events_streams_observer_pattern_java/test.yaml

[coderabbitai]

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Validate longPoll in the builder (non-null and > 0).

longPoll(null) defers failure to a later runtime path, and non-positive durations can cause aggressive empty-page polling in a long-lived subscription loop. Fail fast here.

Proposed fix

+import java.util.Objects;
@@
         public Builder longPoll(Duration longPoll) {
-            this.longPoll = longPoll;
+            Objects.requireNonNull(longPoll, "SubscribeOptions.longPoll is required");
+            if (longPoll.isZero() || longPoll.isNegative()) {
+                throw new IllegalArgumentException(
+                    "SubscribeOptions.longPoll must be > 0 for streaming subscriptions");
+            }
+            this.longPoll = longPoll;
             return this;
         }

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	public Builder longPoll(Duration longPoll) {
	this.longPoll = longPoll;
	return this;
	public Builder longPoll(Duration longPoll) {
	Objects.requireNonNull(longPoll, "SubscribeOptions.longPoll is required");
	if (longPoll.isZero() || longPoll.isNegative()) {
	throw new IllegalArgumentException(
	"SubscribeOptions.longPoll must be > 0 for streaming subscriptions");
	}
	this.longPoll = longPoll;
	return this;
	}

🤖 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 `@src/runtime/java/mcp-mesh-sdk/src/main/java/io/mcpmesh/SubscribeOptions.java`
around lines 109 - 111, In SubscribeOptions.Builder.longPoll, validate the
argument by rejecting null and non-positive durations immediately: in the
Builder.longPoll(Duration longPoll) method (class SubscribeOptions.Builder) call
Objects.requireNonNull(longPoll, "...") and then check that longPoll.toMillis()
> 0 (or longPoll.isZero()/isNegative()) and throw an IllegalArgumentException
with a clear message if invalid; return this unchanged on success. This causes
fast-fail behavior for invalid longPoll values instead of deferring the error to
runtime.