From 5f24acc489bfccc21ed11a0a49711e3e5cfacda0 Mon Sep 17 00:00:00 2001 From: Rui Abreu Date: Sat, 11 Jan 2025 16:10:15 +0000 Subject: [PATCH 1/9] Porting changes from STORM-3693 --- .../src/jvm/org/apache/storm/executor/Executor.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 b3df9c9e631..2d287c77739 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java @@ -285,6 +285,8 @@ 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); } @@ -292,6 +294,10 @@ public void accept(Object event) { 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); From 9227922fa3692802c1d140ab90c200dc537d5301 Mon Sep 17 00:00:00 2001 From: Rui Abreu Date: Sat, 18 Jan 2025 22:31:10 +0000 Subject: [PATCH 2/9] Refactoring changes proposed on STORM-3693 and adding an unit test --- .../org/apache/storm/executor/Executor.java | 10 +-- .../storm/executor/spout/SpoutExecutor.java | 38 ++++++++-- .../storm/executor/SpoutExecutorTest.java | 75 +++++++++++++++++++ 3 files changed, 110 insertions(+), 13 deletions(-) create mode 100644 storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java 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 2d287c77739..562c49102de 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java @@ -285,19 +285,17 @@ 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); } + acceptTupleAction(taskId, tuple); + } + + protected void acceptTupleAction(int taskId, TupleImpl tuple) { 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); diff --git a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java index 25c84474a49..a4081fe6c42 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java @@ -2,9 +2,9 @@ * 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. @@ -17,6 +17,7 @@ import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; + import org.apache.storm.Config; import org.apache.storm.Constants; import org.apache.storm.ICredentialsListener; @@ -80,7 +81,7 @@ public SpoutExecutor(final WorkerState workerData, final List executorId, this.emittedCount = new MutableLong(0); this.emptyEmitStreak = new MutableLong(0); this.stats = new SpoutExecutorStats( - ConfigUtils.samplingRate(this.getTopoConf()), ObjectReader.getInt(this.getTopoConf().get(Config.NUM_STAT_BUCKETS))); + ConfigUtils.samplingRate(this.getTopoConf()), ObjectReader.getInt(this.getTopoConf().get(Config.NUM_STAT_BUCKETS))); this.skippedMaxSpoutMs = workerData.getMetricRegistry().rateCounter("__skipped-max-spout-ms", componentId, taskIds.get(0)); this.skippedInactiveMs = workerData.getMetricRegistry().rateCounter("__skipped-inactive-ms", componentId, @@ -131,8 +132,8 @@ public void expire(Long key, TupleInfo tupleInfo) { } ISpout spoutObject = (ISpout) taskData.getTaskObject(); spoutOutputCollector = new SpoutOutputCollectorImpl( - spoutObject, this, taskData, emittedCount, - hasAckers, rand, hasEventLoggers, isDebug, pending); + spoutObject, this, taskData, emittedCount, + hasAckers, rand, hasEventLoggers, isDebug, pending); SpoutOutputCollector outputCollector = new SpoutOutputCollector(spoutOutputCollector); this.outputCollectors.add(outputCollector); @@ -283,6 +284,29 @@ private void inactiveExecute() throws InterruptedException { skippedInactiveMs.inc(Time.currentTimeMillis() - start); } + @Override + protected void acceptTupleAction(int taskId, TupleImpl tuple) { + + String streamId = tuple.getSourceStreamId(); + + try { + if (taskId != AddressedTuple.BROADCAST_DEST) { + tupleActionFn(taskId, tuple); + } else if (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); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + + } + @Override public void tupleActionFn(int taskId, TupleImpl tuple) throws Exception { String streamId = tuple.getSourceStreamId(); @@ -370,6 +394,6 @@ public int getSpoutRecvqCheckSkipCount() { public long getThreadId() { return threadId; - } - + } + } diff --git a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java new file mode 100644 index 00000000000..d2871f6db10 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java @@ -0,0 +1,75 @@ +package org.apache.storm.executor; + +import org.apache.storm.Constants; +import org.apache.storm.cluster.IStateStorage; +import org.apache.storm.daemon.worker.WorkerState; +import org.apache.storm.executor.spout.SpoutExecutor; +import org.apache.storm.generated.ComponentCommon; +import org.apache.storm.metrics2.RateCounter; +import org.apache.storm.metrics2.StormMetricRegistry; +import org.apache.storm.task.WorkerTopologyContext; +import org.apache.storm.tuple.AddressedTuple; +import org.apache.storm.tuple.TupleImpl; +import org.apache.storm.utils.RotatingMap; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.Test; +import org.junit.platform.commons.util.ReflectionUtils; +import org.mockito.Mockito; + +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; + +public class SpoutExecutorTest { + + + @Test + public void testPendingTuplesRotateShouldBeCalledOnlyOnce() throws Exception { + + RateCounter rateCounter = Mockito.mock(RateCounter.class); + + StormMetricRegistry stormMetricRegistry = Mockito.mock(StormMetricRegistry.class); + Mockito.when(stormMetricRegistry.rateCounter(anyString(),anyString(),anyInt())).thenReturn(rateCounter); + + Map hashmap = Utils.readDefaultConfig(); + + IStateStorage stateStorage = Mockito.mock(IStateStorage.class); + + ComponentCommon componentCommon = Mockito.mock(ComponentCommon.class); + Mockito.when(componentCommon.get_json_conf()).thenReturn(null); + + WorkerTopologyContext workerTopologyContext = Mockito.mock(WorkerTopologyContext.class); + Mockito.when(workerTopologyContext.getComponentId(anyInt())).thenReturn("1"); + Mockito.when(workerTopologyContext.getComponentCommon(anyString())).thenReturn(componentCommon); + + WorkerState workerState = Mockito.mock(WorkerState.class); + Mockito.when(workerState.getWorkerTopologyContext()).thenReturn(workerTopologyContext); + Mockito.when(workerState.getStateStorage()).thenReturn(stateStorage); + Mockito.when(workerState.getTopologyConf()).thenReturn(hashmap); + Mockito.when(workerState.getMetricRegistry()).thenReturn(stormMetricRegistry); + + SpoutExecutor spoutExecutor = new SpoutExecutor(workerState,List.of(1L,5L),new HashMap<>()); + + TupleImpl tuple = Mockito.mock(TupleImpl.class); + Mockito.when(tuple.getSourceStreamId()).thenReturn(Constants.SYSTEM_TICK_STREAM_ID); + AddressedTuple addressedTuple = Mockito.mock(AddressedTuple.class); + Mockito.when(addressedTuple.getDest()).thenReturn(AddressedTuple.BROADCAST_DEST); + Mockito.when(addressedTuple.getTuple()).thenReturn(tuple); + + RotatingMap rotatingMap = Mockito.mock(RotatingMap.class); + Field fieldRotatingMap = ReflectionUtils + .findFields(SpoutExecutor.class, f -> f.getName().equals("pending"), + ReflectionUtils.HierarchyTraversalMode.TOP_DOWN) + .get(0); + fieldRotatingMap.setAccessible(true); + fieldRotatingMap.set(spoutExecutor, rotatingMap); + + spoutExecutor.accept(addressedTuple); + + Mockito.verify(rotatingMap,Mockito.times(1)).rotate(); + } +} From f5db47bc18963a78eebf667ac72f21e68c2b37fc Mon Sep 17 00:00:00 2001 From: reiabreu Date: Sat, 18 Jan 2025 22:44:48 +0000 Subject: [PATCH 3/9] Update Executor.java --- storm-client/src/jvm/org/apache/storm/executor/Executor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 index 562c49102de..edd77743fcd 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java @@ -2,9 +2,9 @@ * 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. From 5ded4cdb67d1463daebe26434571ec09c014ee64 Mon Sep 17 00:00:00 2001 From: reiabreu Date: Sat, 18 Jan 2025 22:45:39 +0000 Subject: [PATCH 4/9] Update SpoutExecutor.java --- .../jvm/org/apache/storm/executor/spout/SpoutExecutor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java index a4081fe6c42..191c1cb0a3b 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java @@ -2,9 +2,9 @@ * 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. From e3f76e60323246c2146e5c1c845c733b8434c042 Mon Sep 17 00:00:00 2001 From: reiabreu Date: Sat, 18 Jan 2025 22:47:12 +0000 Subject: [PATCH 5/9] Update SpoutExecutor.java --- .../jvm/org/apache/storm/executor/spout/SpoutExecutor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java index 191c1cb0a3b..38adeff1872 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java @@ -2,9 +2,9 @@ * 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. From 677756f19ffdaea11d3cc18c7b618e137d59741c Mon Sep 17 00:00:00 2001 From: reiabreu Date: Sat, 18 Jan 2025 22:47:50 +0000 Subject: [PATCH 6/9] Update SpoutExecutor.java --- .../src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java | 1 - 1 file changed, 1 deletion(-) diff --git a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java index 38adeff1872..734fca2a2d6 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java @@ -17,7 +17,6 @@ import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; - import org.apache.storm.Config; import org.apache.storm.Constants; import org.apache.storm.ICredentialsListener; From 365958476695eafdd1cf8ab62a01ecf5bf60a441 Mon Sep 17 00:00:00 2001 From: Rui Abreu Date: Sat, 18 Jan 2025 22:49:57 +0000 Subject: [PATCH 7/9] Adding missing license --- .../apache/storm/executor/SpoutExecutorTest.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java index d2871f6db10..4bcb2fdd6cc 100644 --- a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java +++ b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java @@ -1,3 +1,16 @@ +/** + * 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 org.apache.storm.Constants; From f34416e7f6d99a0fa7224aba0cc37ebbbb6578b1 Mon Sep 17 00:00:00 2001 From: Rui Abreu Date: Sat, 18 Jan 2025 23:26:11 +0000 Subject: [PATCH 8/9] Adding missing license --- .../test/jvm/org/apache/storm/executor/SpoutExecutorTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java index 4bcb2fdd6cc..a44597cde9a 100644 --- a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java +++ b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java @@ -10,7 +10,6 @@ * and limitations under the License. */ - package org.apache.storm.executor; import org.apache.storm.Constants; From 3455c2a9248043d9a1a979da864c4e335b558e8c Mon Sep 17 00:00:00 2001 From: reiabreu Date: Wed, 12 Feb 2025 09:38:21 +0000 Subject: [PATCH 9/9] Update IWorkerHook.java --- .../jvm/org/apache/storm/hooks/IWorkerHook.java | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java b/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java index 7533e1ccd6d..fa60975f2c7 100644 --- a/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java +++ b/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java @@ -22,18 +22,7 @@ * when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down. */ public interface IWorkerHook extends Serializable { - /** - * This method is called when a worker is started. - * - * @param topoConf The Storm configuration for this worker - * @param context This object can be used to get information about this worker's place within the topology - * - * @deprecated see {@link IWorkerHook#start(Map, WorkerUserContext)} - */ - default void start(Map topoConf, WorkerTopologyContext context) { - // NOOP - } - + /** * This method is called when a worker is started and can be used to do necessary prep-processing and allow initialization of shared * application state. @@ -43,7 +32,7 @@ default void start(Map topoConf, WorkerTopologyContext context) * {@link WorkerUserContext#setResource(String, Object)} to set the shared application state. */ default void start(Map topoConf, WorkerUserContext context) { - start(topoConf, context); + // NOOP } /**