From 667210a5198ab8cfc5ef075b4b2fce90fc52cac9 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sat, 4 Jul 2026 01:10:26 +0200 Subject: [PATCH 1/2] decouple the control plane from the data plane on receive queues --- conf/defaults.yaml | 2 + docs/Performance.md | 21 ++ .../src/jvm/org/apache/storm/Config.java | 19 ++ .../src/jvm/org/apache/storm/Constants.java | 46 +++++ .../storm/daemon/worker/WorkerState.java | 15 +- .../org/apache/storm/executor/Executor.java | 49 +++-- .../storm/executor/ExecutorTransfer.java | 8 + .../jvm/org/apache/storm/utils/JCQueue.java | 60 +++++- .../apache/storm/utils/JCQueueMetrics.java | 19 +- .../jvm/org/apache/storm/ConstantsTest.java | 61 ++++++ .../ExecutorTransferControlLaneTest.java | 186 ++++++++++++++++++ .../org/apache/storm/utils/JCQueueTest.java | 172 ++++++++++++++++ 12 files changed, 626 insertions(+), 32 deletions(-) create mode 100644 storm-client/test/jvm/org/apache/storm/ConstantsTest.java create mode 100644 storm-client/test/jvm/org/apache/storm/executor/ExecutorTransferControlLaneTest.java diff --git a/conf/defaults.yaml b/conf/defaults.yaml index d2b1c16eb68..9682cf8bcaf 100644 --- a/conf/defaults.yaml +++ b/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] diff --git a/docs/Performance.md b/docs/Performance.md index 5ca0f4ad62c..e52760a88bf 100644 --- a/docs/Performance.md +++ b/docs/Performance.md @@ -30,6 +30,27 @@ 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. + +- `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. diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 36b3bb78aea..9dceeeed312 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -766,6 +766,25 @@ public class Config extends HashMap { @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. 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. */ diff --git a/storm-client/src/jvm/org/apache/storm/Constants.java b/storm-client/src/jvm/org/apache/storm/Constants.java index 683b4240293..da0a85ee7ee 100644 --- a/storm-client/src/jvm/org/apache/storm/Constants.java +++ b/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,51 @@ 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 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); + + private static final long CONTROL_STREAM_LENGTH_MASK; + private static final long CONTROL_STREAM_THIRD_CHAR_MASK; + + static { + long lengthMask = 0; + long thirdCharMask = 0; + for (String id : SYSTEM_CONTROL_STREAM_IDS) { + if (id.length() < 3 || id.length() >= 64) { + // fail fast + throw new IllegalStateException("control stream id length must be in [3, 63]: " + id); + } + lengthMask |= 1L << id.length(); + thirdCharMask |= 1L << (id.charAt(2) & 63); + } + CONTROL_STREAM_LENGTH_MASK = lengthMask; + CONTROL_STREAM_THIRD_CHAR_MASK = thirdCharMask; + } + + public static boolean isControlStreamId(String streamId) { + // Called per inbound tuple on the ingress hot paths when the control lane is enabled, so avoid + // hashing the string (remote stream ids are freshly deserialized, their hashCode is not cached) + if (streamId == null) { + return false; + } + int len = streamId.length(); + if (len >= 64 || ((CONTROL_STREAM_LENGTH_MASK >>> len) & 1L) == 0) { + return false; + } + if (streamId.charAt(0) != '_' || streamId.charAt(1) != '_' + || ((CONTROL_STREAM_THIRD_CHAR_MASK >>> (streamId.charAt(2) & 63)) & 1L) == 0) { + return false; + } + return 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"; diff --git a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java index 56ac62d53d3..91f24a47c80 100644 --- a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java +++ b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java @@ -571,6 +571,14 @@ private void transferLocalBatch(ArrayList 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, JCQueue> mkReceiveQueueMap(Map 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, JCQueue> mkReceiveQueueMap(Map 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; diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java b/storm-client/src/jvm/org/apache/storm/executor/Executor.java index 59f0894fbe2..d488a626362 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java +++ b/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; } } diff --git a/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java b/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java index ffd25c49440..7052dce4ac4 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java +++ b/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 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); diff --git a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java index 0c26158fad8..8df34b9c418 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java +++ b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java @@ -38,6 +38,8 @@ public class JCQueue implements Closeable { private final MpscArrayQueue recvQueue; // only holds msgs from other workers (via WorkerTransfer), when recvQueue is full private final MpscUnboundedArrayQueue overflowQ; + // dedicated lane for low-volume control tuples (flush/tick/metrics tick/feedback), drained before recvQueue + private final MpscArrayQueue controlQueue; private final int overflowLimit; // ensures... overflowCount <= overflowLimit. if set to 0, disables overflow limiting. private final int producerBatchSz; private final boolean dynamicBatch; @@ -58,15 +60,23 @@ public JCQueue(String queueName, String metricNamePrefix, int size, int overflow public JCQueue(String queueName, String metricNamePrefix, int size, int overflowLimit, int producerBatchSz, IWaitStrategy backPressureWaitStrategy, String topologyId, String componentId, List taskIds, int port, StormMetricRegistry metricRegistry, boolean dynamicBatch) { + this(queueName, metricNamePrefix, size, overflowLimit, producerBatchSz, backPressureWaitStrategy, topologyId, componentId, + taskIds, port, metricRegistry, dynamicBatch, 0); + } + + public JCQueue(String queueName, String metricNamePrefix, int size, int overflowLimit, int producerBatchSz, + IWaitStrategy backPressureWaitStrategy, String topologyId, String componentId, List taskIds, + int port, StormMetricRegistry metricRegistry, boolean dynamicBatch, int controlQueueSize) { this.queueName = queueName; this.dynamicBatch = dynamicBatch; this.overflowLimit = overflowLimit; this.recvQueue = new MpscArrayQueue<>(size); this.overflowQ = new MpscUnboundedArrayQueue<>(size); + this.controlQueue = (controlQueueSize > 0) ? new MpscArrayQueue<>(controlQueueSize) : null; for (Integer taskId : taskIds) { this.jcqMetrics.add(new JCQueueMetrics(metricNamePrefix, topologyId, componentId, taskId, port, - metricRegistry, recvQueue, overflowQ)); + metricRegistry, recvQueue, overflowQ, controlQueue)); } //The batch size can be no larger than half the full recvQueue size, to avoid contention issues. @@ -105,9 +115,14 @@ public int consume(JCQueue.Consumer consumer, ExitCondition exitCond) { } public int size() { - return recvQueue.size() + overflowQ.size(); + int controlCount = (controlQueue == null) ? 0 : controlQueue.size(); + return controlCount + recvQueue.size() + overflowQ.size(); } + /** + * Load of the data plane only. The control lane and overflow lane are deliberately excluded so that load-aware groupings + * (e.g. LoadAwareShuffleGrouping) are unaffected by pending control tuples. + */ public double getQueueLoad() { return ((double) recvQueue.size()) / recvQueue.capacity(); } @@ -116,6 +131,19 @@ public double getQueueLoad() { * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q. */ private int consumeImpl(Consumer consumer, ExitCondition exitCond) throws InterruptedException { + int controlDrainCount = 0; + if (controlQueue != null) { + // drain the control lane first: it is small and bounded, so it cannot starve the data plane + while (exitCond.keepRunning()) { + Object tuple = controlQueue.poll(); + if (tuple == null) { + break; + } + consumer.accept(tuple); + ++controlDrainCount; + } + } + int drainCount = 0; while (exitCond.keepRunning()) { Object tuple = recvQueue.poll(); @@ -133,7 +161,7 @@ private int consumeImpl(Consumer consumer, ExitCondition exitCond) throws Interr ++overflowDrainCount; consumer.accept(tuple); } - int total = drainCount + overflowDrainCount; + int total = controlDrainCount + drainCount + overflowDrainCount; if (total > 0) { consumer.flush(); } @@ -216,6 +244,32 @@ public boolean tryPublishDirect(Object obj) { return tryPublishInternal(obj); } + /** + * Whether this queue has a dedicated control lane for low-volume control tuples. + */ + public boolean isControlLaneEnabled() { + return controlQueue != null; + } + + /** + * Non-blocking, un-batched write to the control lane, exempt from backpressure. Falls back to an un-batched + * write to the recvQueue (same as {@link #tryPublishDirect(Object)}) when the control lane is disabled. + * Returns false if the target queue is full; a failed control publish is dropped by the caller and counted + * via the control drop metric, which is safe because control signals are periodic and self-healing. + */ + public boolean tryPublishControl(Object obj) { + if (controlQueue == null) { + return tryPublishInternal(obj); + } + if (controlQueue.offer(obj)) { + return true; + } + for (JCQueueMetrics jcQueueMetric : jcqMetrics) { + jcQueueMetric.notifyControlMsgDrop(); + } + return false; + } + /** * Un-batched write to overflowQ. Should only be called by WorkerTransfer returns false if overflowLimit has reached */ diff --git a/storm-client/src/jvm/org/apache/storm/utils/JCQueueMetrics.java b/storm-client/src/jvm/org/apache/storm/utils/JCQueueMetrics.java index 6a6a56f4f23..771e13a1b54 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/JCQueueMetrics.java +++ b/storm-client/src/jvm/org/apache/storm/utils/JCQueueMetrics.java @@ -31,10 +31,15 @@ public class JCQueueMetrics implements Closeable { private final RateTracker arrivalsTracker = new RateTracker(10000, 10); private final RateTracker insertFailuresTracker = new RateTracker(10000, 10); private final AtomicLong droppedMessages = new AtomicLong(0); + private final AtomicLong droppedControlMessages = new AtomicLong(0); + /** + * Creates and registers the queue gauges. {@code controlQ} may be null when the queue has no control lane; + * the control-lane gauges are then not registered. + */ public JCQueueMetrics(String metricNamePrefix, String topologyId, String componentId, int taskId, int port, StormMetricRegistry metricRegistry, MpscArrayQueue receiveQ, - MpscUnboundedArrayQueue overflowQ) { + MpscUnboundedArrayQueue overflowQ, MpscArrayQueue controlQ) { Gauge cap = new Gauge() { @Override @@ -103,6 +108,14 @@ public Integer getValue() { metricRegistry.gauge(metricNamePrefix + "-insert_failures", insertFailures, topologyId, componentId, taskId, port); metricRegistry.gauge(metricNamePrefix + "-dropped_messages", dropped, topologyId, componentId, taskId, port); metricRegistry.gauge(metricNamePrefix + "-overflow", overflow, topologyId, componentId, taskId, port); + + if (controlQ != null) { + Gauge controlPop = controlQ::size; + Gauge controlDropped = droppedControlMessages::get; + + metricRegistry.gauge(metricNamePrefix + "-control_population", controlPop, topologyId, componentId, taskId, port); + metricRegistry.gauge(metricNamePrefix + "-control_dropped_messages", controlDropped, topologyId, componentId, taskId, port); + } } public void notifyArrivals(long counts) { @@ -117,6 +130,10 @@ public void notifyDroppedMsg() { droppedMessages.incrementAndGet(); } + public void notifyControlMsgDrop() { + droppedControlMessages.incrementAndGet(); + } + @Override public void close() { arrivalsTracker.close(); diff --git a/storm-client/test/jvm/org/apache/storm/ConstantsTest.java b/storm-client/test/jvm/org/apache/storm/ConstantsTest.java new file mode 100644 index 00000000000..cdf930591ee --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/ConstantsTest.java @@ -0,0 +1,61 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ +package org.apache.storm; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +public class ConstantsTest { + + @Test + public void testEveryWhitelistedStreamIsControl() { + for (String id : Constants.SYSTEM_CONTROL_STREAM_IDS) { + // fresh instance: the fast-reject gates must not depend on identity or a cached hashCode + assertTrue(Constants.isControlStreamId(new String(id.toCharArray())), id); + } + } + + @Test + public void testHighVolumeSystemStreamsAreNotControl() { + // volume-proportional to the data plane: must stay on the data path (see SYSTEM_CONTROL_STREAM_IDS javadoc) + String[] dataPlaneSystemStreams = { + "__ack_init", "__ack_ack", "__ack_fail", "__ack_reset_timeout", + "__metrics", "__system", "__eventlog", "__heartbeat" + }; + for (String id : dataPlaneSystemStreams) { + assertFalse(Constants.isControlStreamId(id), id); + } + } + + @Test + public void testUserStreamsAreNotControl() { + // includes ids whose length collides with a whitelisted id, to exercise the prefix/char gates + String[] userStreams = { "default", "s1", "stream", "__x", "_tick", "___tick", "abcdef", "abcdefghij" }; + for (String id : userStreams) { + assertFalse(Constants.isControlStreamId(id), id); + } + } + + @Test + public void testDegenerateStreamIdsAreNotControl() { + assertFalse(Constants.isControlStreamId(null)); + assertFalse(Constants.isControlStreamId("")); + assertFalse(Constants.isControlStreamId("__")); + StringBuilder longId = new StringBuilder("__"); + for (int i = 0; i < 100; i++) { + longId.append('t'); + } + assertFalse(Constants.isControlStreamId(longId.toString())); + } +} diff --git a/storm-client/test/jvm/org/apache/storm/executor/ExecutorTransferControlLaneTest.java b/storm-client/test/jvm/org/apache/storm/executor/ExecutorTransferControlLaneTest.java new file mode 100644 index 00000000000..4b072be0f53 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/executor/ExecutorTransferControlLaneTest.java @@ -0,0 +1,186 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ +package org.apache.storm.executor; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import org.apache.storm.Constants; +import org.apache.storm.daemon.worker.WorkerState; +import org.apache.storm.generated.StormTopology; +import org.apache.storm.metrics2.StormMetricRegistry; +import org.apache.storm.policy.WaitStrategyPark; +import org.apache.storm.task.GeneralTopologyContext; +import org.apache.storm.task.WorkerTopologyContext; +import org.apache.storm.testing.TestWordCounter; +import org.apache.storm.testing.TestWordSpout; +import org.apache.storm.topology.TopologyBuilder; +import org.apache.storm.tuple.AddressedTuple; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.TupleImpl; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.JCQueue; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Verifies the control-lane routing in {@link ExecutorTransfer#tryTransferLocal}: whitelisted control streams + * (see {@link Constants#SYSTEM_CONTROL_STREAM_IDS}) must reach the destination's control lane regardless of + * pendingEmits ordering and data-queue saturation, while data tuples and the lane-disabled path must keep the + * original semantics. + */ +public class ExecutorTransferControlLaneTest { + + private static final int DEST_TASK_ID = 1; + + private Map topoConf; + private WorkerState workerState; + private GeneralTopologyContext generalTopologyContext; + + @BeforeEach + public void setup() { + topoConf = Utils.readStormConfig(); + + TopologyBuilder builder = new TopologyBuilder(); + builder.setSpout("1", new TestWordSpout(true), 1); + builder.setBolt("2", new TestWordCounter(), 1).fieldsGrouping("1", new Fields("word")); + StormTopology stormTopology = builder.createTopology(); + + WorkerTopologyContext workerTopologyContext = mock(WorkerTopologyContext.class); + when(workerTopologyContext.getRawTopology()).thenReturn(stormTopology); + + workerState = mock(WorkerState.class); + when(workerState.getWorkerTopologyContext()).thenReturn(workerTopologyContext); + generalTopologyContext = mock(GeneralTopologyContext.class); + } + + private ExecutorTransfer mkExecutorTransfer(JCQueue localQueue) { + Map receiveQMap = new HashMap<>(); + receiveQMap.put(DEST_TASK_ID, localQueue); + when(workerState.getLocalReceiveQueues()).thenReturn(receiveQMap); + + ExecutorTransfer executorTransfer = new ExecutorTransfer(workerState, topoConf); + executorTransfer.initLocalRecvQueues(); + return executorTransfer; + } + + private JCQueue mkQueue(String name, int size, int controlQueueSize) { + return new JCQueue(name, name, size, 0, 1, new WaitStrategyPark(100), "test", "test", + Collections.singletonList(DEST_TASK_ID), 6701, new StormMetricRegistry(), false, controlQueueSize); + } + + private AddressedTuple mkTuple(String streamId) { + TupleImpl tuple = new TupleImpl(generalTopologyContext, new Values("v"), Constants.SYSTEM_COMPONENT_ID, + (int) Constants.SYSTEM_TASK_ID, streamId); + return new AddressedTuple(DEST_TASK_ID, tuple); + } + + private static List drain(JCQueue queue) { + List drained = new ArrayList<>(); + queue.consume(new JCQueue.Consumer() { + @Override + public void accept(Object event) { + drained.add(event); + } + + @Override + public void flush() { + } + }); + return drained; + } + + @Test + public void testControlTupleBypassesPendingEmits() { + JCQueue queue = mkQueue("bypassPending", 16, 4); + ExecutorTransfer executorTransfer = mkExecutorTransfer(queue); + + Queue pendingEmits = new ArrayDeque<>(); + pendingEmits.add(mkTuple("default")); // non-empty: the data path would reject and append here + + AddressedTuple controlTuple = mkTuple(Constants.SYSTEM_FLUSH_STREAM_ID); + assertTrue(executorTransfer.tryTransferLocal(controlTuple, queue, pendingEmits), + "control tuple must be reported as handled"); + assertEquals(1, pendingEmits.size(), "control tuple must not be queued behind pendingEmits"); + assertEquals(Collections.singletonList(controlTuple), drain(queue)); + } + + @Test + public void testControlTupleDeliveredWhenDataQueueFull() { + JCQueue queue = mkQueue("fullDataQueue", 16, 4); + ExecutorTransfer executorTransfer = mkExecutorTransfer(queue); + + while (queue.tryPublishDirect("DATA")) { // saturate the data plane + } + + AddressedTuple controlTuple = mkTuple(Constants.SYSTEM_TICK_STREAM_ID); + assertTrue(executorTransfer.tryTransferLocal(controlTuple, queue, null)); + assertEquals(controlTuple, drain(queue).get(0), "control tuple must be drained ahead of the data backlog"); + } + + @Test + public void testControlTupleDroppedOnFullLaneStillReportedHandled() { + JCQueue queue = mkQueue("fullLane", 16, 2); + ExecutorTransfer executorTransfer = mkExecutorTransfer(queue); + + int accepted = 0; + while (accepted < 64 && queue.tryPublishControl("CTRL")) { // saturate the control lane + accepted++; + } + assertTrue(accepted > 0 && accepted < 64); + + Queue pendingEmits = new ArrayDeque<>(); + assertTrue(executorTransfer.tryTransferLocal(mkTuple(Constants.SYSTEM_FLUSH_STREAM_ID), queue, pendingEmits), + "a dropped control tuple is self-healing and must be reported as handled"); + assertTrue(pendingEmits.isEmpty(), "a dropped control tuple must not fall back to pendingEmits"); + assertEquals(accepted, drain(queue).size()); + } + + @Test + public void testDataTupleKeepsPendingEmitsSemantics() { + JCQueue queue = mkQueue("dataPath", 16, 4); + ExecutorTransfer executorTransfer = mkExecutorTransfer(queue); + + Queue pendingEmits = new ArrayDeque<>(); + pendingEmits.add(mkTuple("default")); + + AddressedTuple dataTuple = mkTuple("default"); + assertFalse(executorTransfer.tryTransferLocal(dataTuple, queue, pendingEmits), + "data tuples must keep the original ordering semantics"); + assertEquals(2, pendingEmits.size()); + assertTrue(drain(queue).isEmpty()); + } + + @Test + public void testControlTupleFollowsDataPathWhenLaneDisabled() { + JCQueue queue = mkQueue("laneDisabled", 16, 0); + ExecutorTransfer executorTransfer = mkExecutorTransfer(queue); + + Queue pendingEmits = new ArrayDeque<>(); + pendingEmits.add(mkTuple("default")); + + assertFalse(executorTransfer.tryTransferLocal(mkTuple(Constants.SYSTEM_FLUSH_STREAM_ID), queue, pendingEmits), + "with the lane disabled, control tuples must keep the original data-path semantics"); + assertEquals(2, pendingEmits.size()); + } +} diff --git a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java index e6bf54e1982..28720945195 100644 --- a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java +++ b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java @@ -15,9 +15,15 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.codahale.metrics.Gauge; import java.lang.ref.WeakReference; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; @@ -233,6 +239,154 @@ public void testDynamicBatchGrowsNotShrinksUnderBackpressure() throws Exception assertEquals(3, inserter.batchSize()); } + @Test + public void testControlTuplesDrainedBeforeData() { + JCQueue queue = createQueueWithControlLane("controlFirst", 16, 4); + assertTrue(queue.isControlLaneEnabled()); + + for (int i = 0; i < 8; i++) { + assertTrue(queue.tryPublishDirect(i)); + } + // published AFTER the data backlog, must still be consumed first + assertTrue(queue.tryPublishControl("CTRL")); + + List drained = new ArrayList<>(); + queue.consume(collectingConsumer(drained)); + + assertEquals(9, drained.size()); + assertEquals("CTRL", drained.get(0), "control tuple must be drained ahead of the data backlog"); + } + + @Test + public void testControlLaneDropsOnFullWithoutBlocking() { + JCQueue queue = createQueueWithControlLane("controlDrop", 16, 2); + + int accepted = 0; + while (accepted < 64 && queue.tryPublishControl("CTRL" + accepted)) { + accepted++; + } + assertTrue(accepted > 0 && accepted < 64, "control lane must be bounded"); + + // a full control lane drops (returns false) instead of blocking, and leaves the data path unaffected + assertFalse(queue.tryPublishControl("EXTRA")); + assertTrue(queue.tryPublishDirect("DATA")); + assertEquals(accepted + 1, queue.size()); + } + + @Test + public void testControlPublishFallsBackToRecvQueueWhenLaneDisabled() { + JCQueue queue = createQueue("controlDisabled", 16); + assertFalse(queue.isControlLaneEnabled()); + + assertTrue(queue.tryPublishControl("CTRL")); + assertEquals(1, queue.size()); + + List drained = new ArrayList<>(); + queue.consume(collectingConsumer(drained)); + assertEquals(Collections.singletonList("CTRL"), drained); + } + + @Test + public void testControlTuplesDrainedBeforeOverflow() { + JCQueue queue = createQueueWithControlLane("controlBeforeOverflow", 16, 4); + + assertTrue(queue.tryPublishToOverflow("OVERFLOW")); + assertTrue(queue.tryPublishDirect("DATA")); + assertTrue(queue.tryPublishControl("CTRL")); + + List drained = new ArrayList<>(); + int consumed = queue.consume(collectingConsumer(drained)); + + assertEquals(3, consumed, "consume() return value must include control tuples"); + assertEquals(Arrays.asList("CTRL", "DATA", "OVERFLOW"), drained, + "drain order must be control lane, then recvQueue, then overflowQ"); + } + + @Test + public void testControlLaneIsFifo() { + JCQueue queue = createQueueWithControlLane("controlFifo", 16, 8); + for (int i = 0; i < 5; i++) { + assertTrue(queue.tryPublishControl("CTRL" + i)); + } + + List drained = new ArrayList<>(); + queue.consume(collectingConsumer(drained)); + assertEquals(Arrays.asList("CTRL0", "CTRL1", "CTRL2", "CTRL3", "CTRL4"), drained); + } + + @Test + public void testConsumerFlushCalledAfterControlOnlyDrain() { + JCQueue queue = createQueueWithControlLane("controlFlush", 16, 4); + assertTrue(queue.tryPublishControl("CTRL")); + + AtomicInteger flushCount = new AtomicInteger(); + int consumed = queue.consume(new JCQueue.Consumer() { + @Override + public void accept(Object event) { + } + + @Override + public void flush() { + flushCount.incrementAndGet(); + } + }); + + assertEquals(1, consumed); + assertEquals(1, flushCount.get(), + "a drain that consumed only control tuples must still flush the consumer (e.g. deliver a flush tuple's effect)"); + } + + @Test + public void testExitConditionStopsControlDrain() { + JCQueue queue = createQueueWithControlLane("controlExitCond", 16, 4); + assertTrue(queue.tryPublishControl("CTRL")); + + List drained = new ArrayList<>(); + int consumed = queue.consume(collectingConsumer(drained), () -> false); + + assertEquals(0, consumed, "exit condition must be honored before draining the control lane"); + assertTrue(drained.isEmpty()); + assertEquals(1, queue.size(), "unconsumed control tuple must remain queued"); + } + + @Test + public void testControlDropIsCountedInMetrics() { + StormMetricRegistry registry = new StormMetricRegistry(); + JCQueue queue = new JCQueue("controlDropMetric", "controlDropMetric", 16, 0, 1, waitStrategy, + "test", "test", Collections.singletonList(1000), 1000, registry, false, 2); + + int accepted = 0; + while (accepted < 64 && queue.tryPublishControl("CTRL")) { + accepted++; + } + assertTrue(accepted > 0 && accepted < 64); + assertFalse(queue.tryPublishControl("EXTRA")); // second counted drop (first happened when the fill loop stopped) + + assertEquals((long) accepted, gaugeValue(registry, "control_population"), + "control_population gauge must report the lane occupancy"); + assertEquals(2L, gaugeValue(registry, "control_dropped_messages"), + "every rejected control publish must be counted as a dropped control message"); + } + + private long gaugeValue(StormMetricRegistry registry, String nameSuffix) { + for (Map.Entry entry : registry.getRegistry().getGauges().entrySet()) { + if (entry.getKey().contains(nameSuffix)) { + return ((Number) entry.getValue().getValue()).longValue(); + } + } + throw new AssertionError("gauge not registered: " + nameSuffix); + } + + @Test + public void testQueueLoadExcludesControlLane() { + JCQueue queue = createQueueWithControlLane("controlLoad", 16, 4); + + assertTrue(queue.tryPublishControl("CTRL")); + + assertEquals(0.0, queue.getQueueLoad(), 0.0, "control lane must not contribute to the data-plane load"); + assertEquals(1, queue.size(), "control lane must contribute to size()"); + } + @Test public void testQueueIsCollectedAfterLongLivedProducerPublishes() { Assertions.assertTimeoutPreemptively(Duration.ofSeconds(30), () -> { @@ -325,6 +479,24 @@ private JCQueue createQueue(String name, int batchSize, int queueSize, boolean d new StormMetricRegistry(), dynamicBatch); } + private JCQueue createQueueWithControlLane(String name, int queueSize, int controlQueueSize) { + return new JCQueue(name, name, queueSize, 0, 1, waitStrategy, "test", "test", Collections.singletonList(1000), 1000, + new StormMetricRegistry(), false, controlQueueSize); + } + + private static JCQueue.Consumer collectingConsumer(List sink) { + return new JCQueue.Consumer() { + @Override + public void accept(Object event) { + sink.add(event); + } + + @Override + public void flush() { + } + }; + } + private static class IncProducer implements Runnable { private final JCQueue queue; From 989cd327bb8cd87e0d637bde83979a9866d57a74 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sat, 11 Jul 2026 13:49:49 +0200 Subject: [PATCH 2/2] address review feedback on receive-queue control lane - simplify Constants.isControlStreamId: drop the static-init bitmask fast path in favor of startsWith("__") + set lookup (same behavior, less to maintain) - document that __tick tuples may be dropped under saturation when the lane is enabled (Config javadoc + docs/Performance.md), so bolts using ticks for windowing/expiry tolerate an occasional miss - log a rate-limited WARN in JCQueue.tryPublishControl when a control tuple is dropped, so a stalled executor is visible in the logs and not only via the control_dropped_messages metric - note on the controlQueue field that close() does not drain the lane, matching recvQueue behavior - fix import order in JCQueueTest Co-Authored-By: Claude Fable 5 --- docs/Performance.md | 3 ++ .../src/jvm/org/apache/storm/Config.java | 7 +++- .../src/jvm/org/apache/storm/Constants.java | 35 ++----------------- .../jvm/org/apache/storm/utils/JCQueue.java | 26 +++++++++++++- .../org/apache/storm/utils/JCQueueTest.java | 2 +- 5 files changed, 38 insertions(+), 35 deletions(-) diff --git a/docs/Performance.md b/docs/Performance.md index e52760a88bf..5774a887629 100644 --- a/docs/Performance.md +++ b/docs/Performance.md @@ -46,6 +46,9 @@ saturates. Writes to the control lane are un-batched and never block: if the lan 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 diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 9dceeeed312..f9c09b8973a 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -773,7 +773,12 @@ public class Config extends HashMap { * 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. Default value: false. + * unaffected: they stay on the data path. + * + *

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"; diff --git a/storm-client/src/jvm/org/apache/storm/Constants.java b/storm-client/src/jvm/org/apache/storm/Constants.java index da0a85ee7ee..72abcb74a59 100644 --- a/storm-client/src/jvm/org/apache/storm/Constants.java +++ b/storm-client/src/jvm/org/apache/storm/Constants.java @@ -42,39 +42,10 @@ public class Constants { public static final Set 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); - private static final long CONTROL_STREAM_LENGTH_MASK; - private static final long CONTROL_STREAM_THIRD_CHAR_MASK; - - static { - long lengthMask = 0; - long thirdCharMask = 0; - for (String id : SYSTEM_CONTROL_STREAM_IDS) { - if (id.length() < 3 || id.length() >= 64) { - // fail fast - throw new IllegalStateException("control stream id length must be in [3, 63]: " + id); - } - lengthMask |= 1L << id.length(); - thirdCharMask |= 1L << (id.charAt(2) & 63); - } - CONTROL_STREAM_LENGTH_MASK = lengthMask; - CONTROL_STREAM_THIRD_CHAR_MASK = thirdCharMask; - } - public static boolean isControlStreamId(String streamId) { - // Called per inbound tuple on the ingress hot paths when the control lane is enabled, so avoid - // hashing the string (remote stream ids are freshly deserialized, their hashCode is not cached) - if (streamId == null) { - return false; - } - int len = streamId.length(); - if (len >= 64 || ((CONTROL_STREAM_LENGTH_MASK >>> len) & 1L) == 0) { - return false; - } - if (streamId.charAt(0) != '_' || streamId.charAt(1) != '_' - || ((CONTROL_STREAM_THIRD_CHAR_MASK >>> (streamId.charAt(2) & 63)) & 1L) == 0) { - return false; - } - return SYSTEM_CONTROL_STREAM_IDS.contains(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"; diff --git a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java index 8df34b9c418..5c4c69583c6 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java +++ b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java @@ -22,6 +22,8 @@ import java.lang.ref.WeakReference; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.apache.storm.metrics2.StormMetricRegistry; import org.apache.storm.policy.IWaitStrategy; import org.apache.storm.shade.org.jctools.queues.MessagePassingQueue; @@ -38,7 +40,8 @@ public class JCQueue implements Closeable { private final MpscArrayQueue recvQueue; // only holds msgs from other workers (via WorkerTransfer), when recvQueue is full private final MpscUnboundedArrayQueue overflowQ; - // dedicated lane for low-volume control tuples (flush/tick/metrics tick/feedback), drained before recvQueue + // dedicated lane for low-volume control tuples (flush/tick/metrics tick/feedback), drained before recvQueue. + // Like recvQueue, this lane is not drained on close(): any tuples still buffered at shutdown are discarded. private final MpscArrayQueue controlQueue; private final int overflowLimit; // ensures... overflowCount <= overflowLimit. if set to 0, disables overflow limiting. private final int producerBatchSz; @@ -50,6 +53,10 @@ public class JCQueue implements Closeable { private final IWaitStrategy backPressureWaitStrategy; private final String queueName; + // Throttle for the control-lane-full WARN so a stalled executor cannot flood the logs; log at most once per interval. + private static final long CONTROL_DROP_LOG_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30); + private final AtomicLong lastControlDropLogNanos = new AtomicLong(Long.MIN_VALUE); + public JCQueue(String queueName, String metricNamePrefix, int size, int overflowLimit, int producerBatchSz, IWaitStrategy backPressureWaitStrategy, String topologyId, String componentId, List taskIds, int port, StormMetricRegistry metricRegistry) { @@ -267,9 +274,26 @@ public boolean tryPublishControl(Object obj) { for (JCQueueMetrics jcQueueMetric : jcqMetrics) { jcQueueMetric.notifyControlMsgDrop(); } + maybeWarnControlDrop(); return false; } + // A full control lane means the consuming executor thread is stalled; surface it in the logs (rate-limited) and + // not only via the control_dropped_messages metric, which an operator may not be charting. + private void maybeWarnControlDrop() { + // Reads the clock on every invocation, but this is not a hot path: it runs only when a control tuple is + // dropped, and control tuples reach tryPublishControl only via the isControlStreamId guard (timer-driven + // tick/flush/metrics-tick/feedback signals). So drops are bounded by those signal periods (order of a few + // per second per queue even under a full stall), not by the data rate. A time-based throttle also has to + // read the clock to know the interval elapsed; on Linux nanoTime() is a vDSO read (no syscall/context switch). + long now = System.nanoTime(); + long last = lastControlDropLogNanos.get(); + if (now - last >= CONTROL_DROP_LOG_INTERVAL_NANOS && lastControlDropLogNanos.compareAndSet(last, now)) { + LOG.warn("Control lane full on queue '{}'; dropping control tuple. The consuming executor thread is likely " + + "stalled. See the receive-queue-control_dropped_messages metric.", queueName); + } + } + /** * Un-batched write to overflowQ. Should only be called by WorkerTransfer returns false if overflowLimit has reached */ diff --git a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java index 28720945195..f2b5634734f 100644 --- a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java +++ b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java @@ -23,10 +23,10 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.storm.metrics2.StormMetricRegistry; import org.apache.storm.policy.IWaitStrategy;