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.