Decouple the control plane from the data plane on receive queues

conf/defaults.yaml

@@ -343,6 +343,8 @@ topology.transfer.batch.size: 1       # can be no larger than half of `topology.
 topology.executor.receive.buffer.size: 32768  # size of recv queue for spouts & bolts. Will be internally rounded up to next power of 2 (if not already a power of 2)
 topology.producer.batch.size: 1               # can be no larger than half of `topology.executor.receive.buffer.size`
 topology.producer.batch.dynamic: false        # when true, the producer batch size adapts between 1 and topology.producer.batch.size to reduce latency under light load. No effect unless topology.producer.batch.size > 1.
+topology.executor.receive.control.queue.enable: false  # when true, low-volume control tuples (flush/tick/metrics tick/feedback) are delivered via a dedicated control lane of the recv queue, drained before the data queue and exempt from backpressure.
+topology.executor.receive.control.buffer.size: 1024    # size of the control lane of the recv queue. Only used when topology.executor.receive.control.queue.enable is true. Will be internally rounded up to next power of 2 (if not already a power of 2)
 
 topology.batch.flush.interval.millis: 1  # Flush tuples are disabled if this is set to 0 or if (topology.producer.batch.size=1 and topology.transfer.batch.size=1).
 topology.spout.recvq.skips: 3  # Check recvQ once every N invocations of Spout's nextTuple() [when ACKs disabled]

docs/Performance.md

@@ -30,6 +30,30 @@ Very large message queues are also not desirable to deal with slow consumers. Be
 are large and often full, the messages will end up waiting longer in these queues at each step of the processing, leading to poor latency being
 reported on the Storm UI. Large queues also imply higher memory consumption especially if the queues are typically full.
 
+#### Control lane for system control tuples
+
+An executor's receive queue carries two very different kinds of traffic: the data tuples flowing between components, and low-volume time-driven
+system tuples (flush tuples, tick tuples, metrics ticks, upstream feedback) that keep the topology responsive. Because the queue is FIFO, a control
+tuple enqueued behind a full data backlog waits for the entire backlog to drain first, so data-plane saturation directly degrades control-plane
+latency.
+
+- `topology.executor.receive.control.queue.enable` (default `false`) : When enabled, each executor's receive queue gets a small dedicated *control
+lane* that is drained before the data queue on every consume pass. The time-driven system streams (`__flush`, `__tick`, `__metrics_tick`,
+`__feedback_tick`, `__feedback`) are routed to it at every ingress point (timer-generated tuples, intra-worker transfer, and inter-worker receive),
+insulating them from data-plane buffering, backpressure and overflow so that their delivery latency stays bounded even while the data plane
+saturates. Writes to the control lane are un-batched and never block: if the lane is full the tuple is dropped and counted (see the
+`receive-queue-control_dropped_messages` metric), which is safe because these signals are periodic and the next one arrives within the signal's
+period. The ack streams and the `__metrics` payload stream are volume-proportional to the data plane and stay on the data path, so acking,
+`max.spout.pending` throttling and at-least-once semantics are unaffected. Note that enabling this delivers control tuples ahead of co-enqueued data
+tuples; the whitelisted streams are wall-clock signals with no ordering contract against data, so this is safe.
+Note also that `__tick` is one of these control streams, so the drop-when-full behavior applies to user-facing tick tuples too: with the lane
+enabled a `__tick` tuple may occasionally be dropped under sustained saturation instead of blocking until delivered. Bolts that rely on tick tuples
+for windowing or expiry logic must tolerate an occasional missed tick — it is not only internal signals that can be dropped.
+
+- `topology.executor.receive.control.buffer.size` (default `1024`) : The size of the control lane. Control traffic is low-volume, so the default is
+ample; like the other queue sizes it is internally rounded up to the next power of 2. The control lane is excluded from the queue load reported to
+load-aware groupings.
+
 
 ## 2. Batch Size
 Producers can either write a batch of messages to the consumer's queue or write each message individually. This batch size can be configured.

storm-client/src/jvm/org/apache/storm/Config.java

@@ -766,6 +766,30 @@ public class Config extends HashMap<String, Object> {
     @IsPositiveNumber
     @IsInteger
     public static final String TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE = "topology.executor.receive.buffer.size";
+    /**
+     * When enabled, each executor's receive queue gets a small, dedicated control lane that is drained before the
+     * data queue. Low-volume time-driven system tuples (specified here {@link Constants#SYSTEM_CONTROL_STREAM_IDS}) are routed to
+     * it, insulating them from data-plane buffering, backpressure and overflow so their delivery latency stays
+     * bounded while the data plane saturates. Control tuples are published to the lane without batching and without
+     * participating in backpressure; if the lane is full they are dropped (and counted), which is safe because these
+     * signals are periodic and the next one arrives within the signal's period. Ack and metrics payload streams are
+     * unaffected: they stay on the data path.
+     *
+     * <p>Note that {@code __tick} tuples travel on the control lane. When this is enabled a {@code __tick} tuple may
+     * occasionally be dropped under sustained saturation rather than blocking until delivered as it does with the lane
+     * off. Bolts that rely on tick tuples for windowing or expiry logic must therefore tolerate an occasional missed
+     * tick; it is not only internal signals that may be dropped. Default value: false.
+     */
+    @IsBoolean
+    public static final String TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_QUEUE_ENABLE = "topology.executor.receive.control.queue.enable";
+    /**
+     * The size of the control lane of the receive queue for each executor. Only used when {@link
+     * #TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_QUEUE_ENABLE} is set. Control traffic is low-volume, so a small buffer
+     * suffices; will be internally rounded up to the next power of 2. Default value: 1024.
+     */
+    @IsPositiveNumber
+    @IsInteger
+    public static final String TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_BUFFER_SIZE = "topology.executor.receive.control.buffer.size";
     /**
      * The size of the transfer queue for each worker.
      */

storm-client/src/jvm/org/apache/storm/Constants.java

@@ -14,6 +14,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Set;
 import org.apache.storm.coordination.CoordinatedBolt;
 
 
@@ -31,6 +32,22 @@ public class Constants {
     public static final String FEEDBACK_STREAM_ID = "__feedback";
     public static final String FEEDBACK_TICK_STREAM_ID = "__feedback_tick";
 
+    /**
+     * System streams carrying low-volume, time-driven control signals that may be routed to the executor receive
+     * queue's control lane (see {@code Config.TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_QUEUE_ENABLE}). Tuples on these
+     * streams are wall-clock signals with no ordering contract against data tuples, so delivering them ahead of
+     * co-enqueued data tuples is safe, and dropping one when the control lane is full is self-healing because the
+     * next one arrives within the signal's period.
+     */
+    public static final Set<String> SYSTEM_CONTROL_STREAM_IDS =
+            Set.of(SYSTEM_TICK_STREAM_ID, SYSTEM_FLUSH_STREAM_ID, METRICS_TICK_STREAM_ID, FEEDBACK_STREAM_ID, FEEDBACK_TICK_STREAM_ID);
+
+    public static boolean isControlStreamId(String streamId) {
+        // All control stream ids start with "__"; the prefix check cheaply rejects the common data-stream
+        // case before falling through to the set lookup.
+        return streamId != null && streamId.startsWith("__") && SYSTEM_CONTROL_STREAM_IDS.contains(streamId);
+    }
+
     public static final Object TOPOLOGY = "topology";
     public static final String SYSTEM_TOPOLOGY = "system-topology";
     public static final String STORM_CONF = "storm-conf";

storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java

@@ -571,6 +571,14 @@ private void transferLocalBatch(ArrayList<AddressedTuple> tupleBatch) {
             AddressedTuple tuple = tupleBatch.get(i);
             JCQueue queue = taskToExecutorQueue.get(tuple.dest);
 
+            // 0- route control tuples to the control lane: un-batched, ahead of any data backlog, exempt from
+            //    backpressure and overflow. If the lane is full the tuple is dropped and counted; control signals
+            //    are periodic, so the next one arrives within its period.
+            if (queue.isControlLaneEnabled() && Constants.isControlStreamId(tuple.getTuple().getSourceStreamId())) {
+                queue.tryPublishControl(tuple);
+                continue;
+            }
+
             // 1- try adding to main queue if its overflow is not empty
             if (queue.isEmptyOverflow()) {
                 if (queue.tryPublish(tuple)) {
@@ -726,6 +734,11 @@ private Map<List<Long>, JCQueue> mkReceiveQueueMap(Map<String, Object> topologyC
         Integer recvBatchSize = ObjectReader.getInt(topologyConf.get(Config.TOPOLOGY_PRODUCER_BATCH_SIZE));
         boolean dynamicBatch = ObjectReader.getBoolean(topologyConf.get(Config.TOPOLOGY_PRODUCER_BATCH_DYNAMIC), false);
         Integer overflowLimit = ObjectReader.getInt(topologyConf.get(Config.TOPOLOGY_EXECUTOR_OVERFLOW_LIMIT));
+        boolean controlQueueEnable =
+            ObjectReader.getBoolean(topologyConf.get(Config.TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_QUEUE_ENABLE), false);
+        int controlQueueSize = controlQueueEnable
+            ? ObjectReader.getInt(topologyConf.get(Config.TOPOLOGY_EXECUTOR_RECEIVE_CONTROL_BUFFER_SIZE), 1024)
+            : 0;
 
         if (recvBatchSize > recvQueueSize / 2) {
             throw new IllegalArgumentException(Config.TOPOLOGY_PRODUCER_BATCH_SIZE + ":" + recvBatchSize
@@ -747,7 +760,7 @@ private Map<List<Long>, JCQueue> mkReceiveQueueMap(Map<String, Object> topologyC
             }
             receiveQueueMap.put(executor, new JCQueue("receive-queue" + executor.toString(), "receive-queue",
                                                       recvQueueSize, overflowLimit, recvBatchSize, backPressureWaitStrategy,
-                this.getTopologyId(), compId, taskIds, this.getPort(), metricRegistry, dynamicBatch));
+                this.getTopologyId(), compId, taskIds, this.getPort(), metricRegistry, dynamicBatch, controlQueueSize));
 
         }
         return receiveQueueMap;

storm-client/src/jvm/org/apache/storm/executor/Executor.java

@@ -577,17 +577,24 @@ private void scheduleMetricsTick(int interval) {
                         new TupleImpl(workerTopologyContext, new Values(interval), Constants.SYSTEM_COMPONENT_ID,
                                 (int) Constants.SYSTEM_TASK_ID, Constants.METRICS_TICK_STREAM_ID);
                 AddressedTuple metricsTickTuple = new AddressedTuple(AddressedTuple.BROADCAST_DEST, tuple);
-                try {
-                    receiveQueue.publish(metricsTickTuple);
-                    receiveQueue.flush();  // avoid buffering
-                } catch (InterruptedException e) {
-                    LOG.warn("Thread interrupted when publishing metrics. Setting interrupt flag.");
-                    Thread.currentThread().interrupt();
-                    return;
-                }
+                publishTimerTuple(metricsTickTuple, "metrics tick tuple");
             }
         );
     }
+
+    private void publishTimerTuple(AddressedTuple timerTuple, String logContext) {
+        if (receiveQueue.isControlLaneEnabled()) {
+            receiveQueue.tryPublishControl(timerTuple);
+            return;
+        }
+        try {
+            receiveQueue.publish(timerTuple);
+            receiveQueue.flush();  // avoid buffering
+        } catch (InterruptedException e) {
+            LOG.warn("Thread interrupted when publishing {}. Setting interrupt flag.", logContext);
+            Thread.currentThread().interrupt();
+        }
+    }
 
     /**
      * Collects the task ids of every upstream (source component) task. These are the recipients of
@@ -618,14 +625,7 @@ protected void scheduleUpstreamFeedbackTick(int interval) {
                         new TupleImpl(workerTopologyContext, new Values(interval), Constants.SYSTEM_COMPONENT_ID,
                                 (int) Constants.SYSTEM_TASK_ID, Constants.FEEDBACK_TICK_STREAM_ID);
                 AddressedTuple feedbackTickTuple = new AddressedTuple(AddressedTuple.BROADCAST_DEST, tuple);
-                try {
-                    receiveQueue.publish(feedbackTickTuple);
-                    receiveQueue.flush();  // avoid buffering
-                } catch (InterruptedException e) {
-                    LOG.warn("Thread interrupted when publishing upstream feedback tick. Setting interrupt flag.");
-                    Thread.currentThread().interrupt();
-                    return;
-                }
+                publishTimerTuple(feedbackTickTuple, "upstream feedback tick tuple");
             }
         );
     }
@@ -663,14 +663,7 @@ protected void setupTicks(boolean isSpout) {
                                                         (int) Constants.SYSTEM_TASK_ID,
                                                         Constants.SYSTEM_TICK_STREAM_ID);
                         AddressedTuple tickTuple = new AddressedTuple(AddressedTuple.BROADCAST_DEST, tuple);
-                        try {
-                            receiveQueue.publish(tickTuple);
-                            receiveQueue.flush(); // avoid buffering
-                        } catch (InterruptedException e) {
-                            LOG.warn("Thread interrupted when emitting tick tuple. Setting interrupt flag.");
-                            Thread.currentThread().interrupt();
-                            return;
-                        }
+                        publishTimerTuple(tickTuple, "tick tuple");
                     }
                 );
             }
@@ -683,13 +676,15 @@ public void reflectNewLoadMapping(LoadMapping loadMapping) {
         }
     }
 
-    // Called by flush-tuple-timer thread
+    // Called by flush-tuple-timer thread. Publishes to the control lane when enabled (insulating the flush signal
+    // from the data backlog); otherwise falls back to an un-batched write to the recvQueue as before.
     public boolean publishFlushTuple() {
-        if (receiveQueue.tryPublishDirect(flushTuple)) {
+        if (receiveQueue.tryPublishControl(flushTuple)) {
             LOG.debug("Published Flush tuple to: {} ", getComponentId());
             return true;
         } else {
-            LOG.debug("RecvQ is currently full, will retry publishing Flush Tuple later to : {}", getComponentId());
+            LOG.debug("Target queue (control lane or RecvQ) is currently full, will retry publishing Flush Tuple later to : {}",
+                getComponentId());
             return false;
         }
     }

storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java

@@ -17,6 +17,7 @@
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 import org.apache.storm.Config;
+import org.apache.storm.Constants;
 import org.apache.storm.daemon.worker.WorkerState;
 import org.apache.storm.serialization.KryoTupleSerializer;
 import org.apache.storm.task.WorkerTopologyContext;
@@ -99,6 +100,13 @@ public JCQueue getLocalQueue(AddressedTuple tuple) {
      */
     public boolean tryTransferLocal(AddressedTuple tuple, JCQueue localQueue, Queue<AddressedTuple> pendingEmits) {
         workerData.checkSerialize(threadLocalSerializer.get(), tuple);
+        if (localQueue.isControlLaneEnabled() && Constants.isControlStreamId(tuple.getTuple().getSourceStreamId())) {
+            // control tuples bypass batching, pendingEmits ordering and backpressure. If the control lane is full
+            // the tuple is dropped and counted; control signals are periodic, so the next one arrives within its
+            // period. Report the tuple as handled either way so it is never queued behind the data backlog.
+            localQueue.tryPublishControl(tuple);
+            return true;
+        }
         if (pendingEmits != null) {
             if (pendingEmits.isEmpty() && localQueue.tryPublish(tuple)) {
                 queuesToFlush.set(tuple.dest - indexingBase, localQueue);