Porting changes from STORM-3693

apache · Jan 11, 2025 · 5f24acc · 5f24acc
1 parent 457194d
commit 5f24acc
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
@@ -285,13 +285,19 @@ public void accept(Object event) {
         int taskId = addressedTuple.getDest();
 
         TupleImpl tuple = (TupleImpl) addressedTuple.getTuple();
+        String streamId = tuple.getSourceStreamId();
+        boolean isSpout = this instanceof SpoutExecutor;
         if (isDebug) {
             LOG.info("Processing received TUPLE: {} for TASK: {} ", tuple, taskId);
         }
 
         try {
             if (taskId != AddressedTuple.BROADCAST_DEST) {
                 tupleActionFn(taskId, tuple);
+            } else if (isSpout && streamId.equals(Constants.SYSTEM_TICK_STREAM_ID)) {
+                //taskId is irrelevant here. Ensures pending.rotate() is called once per tick.
+                tupleActionFn(taskIds.get(0), tuple);
+
             } else {
                 for (Integer t : taskIds) {
                     tupleActionFn(t, tuple);