Decouple the control plane from the data plane on receive queues

[GGraziadei]

What is the purpose of the change

Fixes #8816

A Storm executor multiplexes two fundamentally different kinds of traffic onto its single inbound JCQueue:

• Data plane: the actual business tuples flowing between components. High volume, bursty, and the
  traffic that backpressure is meant to throttle.
• Control plane: low-volume, coordination tuples that keep the topology healthy and responsive.

This PR introduces an optional control receive queue. When enabled, this dedicated queue is prioritized and completely drained before the standard data queue is processed. This ensures that critical coordination traffic bypasses data plane bottlenecks.
For additional context and technical details, please refer to the discussion inside the issue thread.

How was the change tested

• Unit tests
• Benchmarcks (a report will be available in the discussion asap)

[GGraziadei]

Thanks @reiabreu, your intuition was entirely correct. Using a streaming grouping driven by a feedback control loop is fundamental to decoupling the control plane from the data plane. In particular, enabling JitterAwareStreamingGroup with a dedicated control queue (CQ) is a game-changer (details are in the attached report).

Here are the key takeaways from the performance scenarios:

• Under No-Backpressure Conditions (No Jitter Asymmetry) FileReadWordCountTopo: Enabling a dedicated control queue introduces an unnecessary overhead. Because the single-threaded implementation must poll the control queue before the data queue, data plane performance is penalized linearly relative to traffic volume in control plane (e.g., metrics or ticks). In this scenario, the control queue provides no actionable data to optimize the data plane.
• Under Backpressured Conditions: Enabling the control queue acts as an invariant. The system's performance bottleneck is driven entirely by the backpressure itself, rather than the minor overhead of the thread polling the control queue.

In conclusion, it makes sense to keep the dedicated control queue (CQ) optional. It should not be enabled by default.

Jitter-Aware Stream Grouping_ Decoupling the Control Queue.pdf

Evaluating the Effect of the Control Queue in a Backpressured Scenario.pdf

Evaluating the cost of a separate Control Queue.pdf

Raw data:
BackPressureTopo.txt
FileReadWordCountTopo.txt
JitterAwareGroupingTopology.txt

[rzo1]

Thx for the PR. The overall design looks good to me: opt-in, default off and well tested. A few things I would like to see changed before merging:

1. Please simplify Constants.isControlStreamId. The hand-rolled bitmask fast path is a lot of subtle static-init machinery to avoid a single Set.contains on a short string. Unless a benchmark showed this matters, a plain streamId != null && streamId.startsWith("__") && SYSTEM_CONTROL_STREAM_IDS.contains(streamId) is easier to maintain. If you keep the fast path, please add a short note on why it is needed.
2. Document the tick tuple drop semantics in the Config javadoc and in docs/Performance.md. With the lane enabled, a full lane silently drops a __tick tuple that previously would have blocked until delivered. User bolts can rely on tick tuples for windowing or expiry logic, so please state explicitly that user-facing tick tuples may occasionally be dropped under saturation, not just internal signals.
3. Add a rate-limited WARN log when a control tuple is dropped in tryPublishControl. A nonzero control_dropped_messages means the executor thread is stalled, which operators should see in the logs and not only if they happen to chart the new gauge.
4. Please add a short note (javadoc on the new constructor or on the field) that close() intentionally does not drain the control lane, matching the existing recvQueue behavior, so this is documented rather than implicit.
5. Nit: the new imports in JCQueueTest are slightly out of order (java.util.concurrent.atomic.AtomicInteger before java.util.concurrent.ExecutorService), checkstyle may flag this.

The remaining behavioral risks (control tuples overtaking co-enqueued data, drop instead of block) are fine with me as long as they are documented as per point 2.

[GGraziadei]

Hi @rzo1,
Thanks for the detailed review and the feedback!
First, just for full transparency, I wanted to highlight that the overall design reflects what we discussed in the dedicated thread, where you provided a key contribution to shaping this approach.

I have just pushed a commit that addresses all your points. Interestingly, to address most of the documentation, logging, and nitpick fixes, I ran an experiment and let Claude handle them directly. I'm sure you're already familiar with these tools, but the agent did a good job. It feels like projects like Apache Magpie might really be just around the corner?

On my end, I focused on the micro-benchmarking to see if the bitmask for isControlStreamId was truly justified on the hot path.

Here are the results of the JMH microbenchmark (2 forks, 10 measured iterations per case, JDK 25, jvm warmup discarded) comparing the bitmask against the simplified startsWith("__") + Set.contains approach across representative stream ids:

• Data streams: Both implementations perform identically (~1 ns cached / ~4 ns fresh) because both quickly reject the string on the very first character without hashing.
• Control streams (e.g., __tick): Performance is identical (~5 ns cached / ~11 ns fresh) since both eventually route to the same Set.contains lookup.
• Edge case: The bitmask only wins in the rare scenario of __-prefixed non-control IDs that miss the mask (e.g., __foo is ~1.5 ns faster cached / ~4.9 ns faster fresh by skipping the hash). However, it is actually slower if those filters don't catch the string and it has to hash anyway (e.g., __fabcd: 6.6 vs 5.7 ns cached, 12.9 vs 12.2 ns fresh).

Even if we consider a worst-case scenario where a sequence of these edge-case IDs adds around 5 ns per check, across a standard path size of 10 to 100 hops (considering the order of magnitude), the cumulative latency increase would be roughly ~500 ns. Given that per-tuple processing overhead is measured in microseconds, this is completely negligible. Since real topologies are dominated by user data streams and genuine control tuples, the ~30 lines of fragile static-init machinery are optimizing a case that doesn't realistically occur.

I completely agree that simplifying the code and using a standard HashSet is the right call. The clean-up is included in the latest commit.

[reiabreu]

@GGraziadei thanks for this great work and for the detailed performance reports.