This is an automated email from the ASF dual-hosted git repository.
GGraziadei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 9dc65e94a Decouple the control plane from the data plane on receive
queues (#8892)
9dc65e94a is described below
commit 9dc65e94aa9438caf6a576bed1f68d8f0760eb57
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Wed Jul 15 20:04:03 2026 +0200
Decouple the control plane from the data plane on receive queues (#8892)
* decouple the control plane from the data plane on receive queues
* 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
---
conf/defaults.yaml | 2 +
docs/Performance.md | 24 +++
storm-client/src/jvm/org/apache/storm/Config.java | 24 +++
.../src/jvm/org/apache/storm/Constants.java | 17 ++
.../apache/storm/daemon/worker/WorkerState.java | 15 +-
.../jvm/org/apache/storm/executor/Executor.java | 49 +++---
.../apache/storm/executor/ExecutorTransfer.java | 8 +
.../src/jvm/org/apache/storm/utils/JCQueue.java | 84 +++++++++-
.../jvm/org/apache/storm/utils/JCQueueMetrics.java | 19 ++-
.../test/jvm/org/apache/storm/ConstantsTest.java | 61 +++++++
.../executor/ExecutorTransferControlLaneTest.java | 186 +++++++++++++++++++++
.../jvm/org/apache/storm/utils/JCQueueTest.java | 172 +++++++++++++++++++
12 files changed, 629 insertions(+), 32 deletions(-)
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index d2b1c16eb..9682cf8bc 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 5ca0f4ad6..5774a8876 100644
--- a/docs/Performance.md
+++ b/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.
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 36b3bb78a..f9c09b897 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/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.
*/
diff --git a/storm-client/src/jvm/org/apache/storm/Constants.java
b/storm-client/src/jvm/org/apache/storm/Constants.java
index 683b42402..72abcb74a 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 @@ package org.apache.storm;
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";
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 496c503e9..59aceb8d6 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
@@ -578,6 +578,14 @@ public class WorkerState {
continue;
}
+ // 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)) {
@@ -739,6 +747,11 @@ public class WorkerState {
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
@@ -760,7 +773,7 @@ public class WorkerState {
}
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 59f0894fb..d488a6263 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 @@ public abstract class Executor implements Callable,
JCQueue.Consumer {
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 @@ public abstract class Executor implements Callable,
JCQueue.Consumer {
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 @@ public abstract class Executor implements Callable,
JCQueue.Consumer {
(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 abstract class Executor implements Callable,
JCQueue.Consumer {
}
}
- // 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 ffd25c494..7052dce4a 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.Map;
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 class ExecutorTransfer {
*/
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);
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 0c26158fa..5c4c69583 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.io.Closeable;
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,6 +40,9 @@ public class JCQueue implements Closeable {
private final MpscArrayQueue<Object> recvQueue;
// only holds msgs from other workers (via WorkerTransfer), when recvQueue
is full
private final MpscUnboundedArrayQueue<Object> overflowQ;
+ // 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<Object> controlQueue;
private final int overflowLimit; // ensures... overflowCount <=
overflowLimit. if set to 0, disables overflow limiting.
private final int producerBatchSz;
private final boolean dynamicBatch;
@@ -48,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<Integer> taskIds,
int port, StormMetricRegistry metricRegistry) {
@@ -58,15 +67,23 @@ public class JCQueue implements Closeable {
public JCQueue(String queueName, String metricNamePrefix, int size, int
overflowLimit, int producerBatchSz,
IWaitStrategy backPressureWaitStrategy, String topologyId,
String componentId, List<Integer> 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<Integer> 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 +122,14 @@ public class JCQueue implements Closeable {
}
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 +138,19 @@ public class JCQueue implements Closeable {
* 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 +168,7 @@ public class JCQueue implements Closeable {
++overflowDrainCount;
consumer.accept(tuple);
}
- int total = drainCount + overflowDrainCount;
+ int total = controlDrainCount + drainCount + overflowDrainCount;
if (total > 0) {
consumer.flush();
}
@@ -216,6 +251,49 @@ public class JCQueue implements Closeable {
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();
+ }
+ 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/src/jvm/org/apache/storm/utils/JCQueueMetrics.java
b/storm-client/src/jvm/org/apache/storm/utils/JCQueueMetrics.java
index 6a6a56f4f..771e13a1b 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<Object> receiveQ,
- MpscUnboundedArrayQueue<Object> overflowQ) {
+ MpscUnboundedArrayQueue<Object> overflowQ,
MpscArrayQueue<Object> controlQ) {
Gauge<Integer> cap = new Gauge<Integer>() {
@Override
@@ -103,6 +108,14 @@ public class JCQueueMetrics implements Closeable {
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<Integer> controlPop = controlQ::size;
+ Gauge<Long> 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 class JCQueueMetrics implements Closeable {
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 000000000..cdf930591
--- /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 000000000..4b072be0f
--- /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<String, Object> 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<Integer, JCQueue> 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<Object> drain(JCQueue queue) {
+ List<Object> 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<AddressedTuple> 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<AddressedTuple> 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<AddressedTuple> 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<AddressedTuple> 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 e6bf54e19..f2b563473 100644
--- a/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java
+++ b/storm-client/test/jvm/org/apache/storm/utils/JCQueueTest.java
@@ -15,12 +15,18 @@ 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 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.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;
@@ -233,6 +239,154 @@ public class JCQueueTest {
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<Object> 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<Object> 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<Object> 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<Object> 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<Object> 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<String, Gauge> 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 @@ public class JCQueueTest {
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<Object> 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;
