You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by rz...@apache.org on 2023/08/25 10:18:14 UTC
[storm] branch master updated (939d92c6d -> 96cc304b0)
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
from 939d92c6d Fix cli monitor component's argument type
new 45d141cc2 support worker hook declaration in flux
new 374eee25c updating flux README
new f75ab7853 minor correction in flux README
new 96cc304b0 update README
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/flux.md | 15 ++++++++
.../src/main/java/org/apache/storm/flux/Flux.java | 6 +++
.../java/org/apache/storm/flux/FluxBuilder.java | 17 +++++++++
.../org/apache/storm/flux/model/TopologyDef.java | 43 ++++++++++++++++++++++
.../model/{BoltDef.java => WorkerHookDef.java} | 4 +-
.../test/java/org/apache/storm/flux/TCKTest.java | 13 +++++++
...d_static_factory_test.yaml => worker_hook.yaml} | 21 +++--------
7 files changed, 102 insertions(+), 17 deletions(-)
copy flux/flux-core/src/main/java/org/apache/storm/flux/model/{BoltDef.java => WorkerHookDef.java} (90%)
copy flux/flux-core/src/test/resources/configs/{bad_static_factory_test.yaml => worker_hook.yaml} (84%)
[storm] 02/04: updating flux README
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 374eee25cac240bc5567a579eda8232ad35766b6
Author: Bharat Gulati <bh...@gmail.com>
AuthorDate: Sat Jun 17 14:18:45 2023 +0530
updating flux README
---
docs/flux.md | 15 +++++++++++++++
1 file changed, 15 insertions(+)
diff --git a/docs/flux.md b/docs/flux.md
index cf83741c4..75a62e133 100644
--- a/docs/flux.md
+++ b/docs/flux.md
@@ -244,6 +244,7 @@ definition consists of the following:
* A list of spouts, each identified by a unique ID
* A list of bolts, each identified by a unique ID
* A list of "stream" objects representing a flow of tuples between spouts and bolts
+ * (Optional) A list of "workerHooks", each identifed by a unique ID
4. **OR** (A JVM class that can produce a `org.apache.storm.generated.StormTopology` instance:
* A `topologySource` definition.
@@ -286,6 +287,10 @@ streams:
grouping:
type: SHUFFLE
+# worker hook definitions
+workerHooks:
+ - id: "base-worker-hook"
+ className: "org.apache.storm.hooks.BaseWorkerHook"
```
## Property Substitution/Filtering
@@ -765,6 +770,16 @@ The `override` property controls how includes affect the values defined in the c
**N.B.:** Includes are not yet recursive. Includes from included files will be ignored.
+## Worker Hooks
+Flux allows you to attach topology component that can be executed when a worker starts, and when a worker shuts down. It can be useful when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down, e.g. managing application context. Worker Hooks should be an implementation of [IWorkerHook]({{page.git-blob-base}}/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java). Otherwise, follow similar Bean definition semantics as [...]
+
+Worker Hooks are specified as a list of maps:
+
+```yaml
+workerHooks:
+ - id: "base-worker-hook"
+ className: "org.apache.storm.hooks.BaseWorkerHook"
+```
## Basic Word Count Example
[storm] 01/04: support worker hook declaration in flux
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 45d141cc24b9c5b65a90fbb6667140350336f84b
Author: Bharat Gulati <bh...@gmail.com>
AuthorDate: Sat Jun 17 13:58:48 2023 +0530
support worker hook declaration in flux
---
.../src/main/java/org/apache/storm/flux/Flux.java | 6 +++
.../java/org/apache/storm/flux/FluxBuilder.java | 17 ++++++++
.../org/apache/storm/flux/model/TopologyDef.java | 43 ++++++++++++++++++++
.../org/apache/storm/flux/model/WorkerHookDef.java | 25 ++++++++++++
.../test/java/org/apache/storm/flux/TCKTest.java | 13 ++++++
.../src/test/resources/configs/worker_hook.yaml | 47 ++++++++++++++++++++++
6 files changed, 151 insertions(+)
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java b/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
index 15f492d10..c89c4d70c 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
@@ -39,6 +39,7 @@ import org.apache.storm.flux.model.ExecutionContext;
import org.apache.storm.flux.model.SpoutDef;
import org.apache.storm.flux.model.StreamDef;
import org.apache.storm.flux.model.TopologyDef;
+import org.apache.storm.flux.model.WorkerHookDef;
import org.apache.storm.flux.parser.FluxParser;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.SubmitOptions;
@@ -210,6 +211,11 @@ public class Flux {
for (StreamDef sd : t.getStreams()) {
printf("%s --%s--> %s", sd.getFrom(), sd.getGrouping().getType(), sd.getTo());
}
+
+ print("--------------- WORKER HOOKS ---------------");
+ for (WorkerHookDef whd : t.getWorkerHooks()) {
+ printf("%s (%s)", whd.getId(), whd.getClassName());
+ }
print("--------------------------------------");
}
}
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java b/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
index 3e5fbe0d9..148852ebf 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
@@ -43,8 +43,10 @@ import org.apache.storm.flux.model.PropertyDef;
import org.apache.storm.flux.model.SpoutDef;
import org.apache.storm.flux.model.StreamDef;
import org.apache.storm.flux.model.TopologyDef;
+import org.apache.storm.flux.model.WorkerHookDef;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.hooks.IWorkerHook;
import org.apache.storm.topology.BoltDeclarer;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
@@ -117,6 +119,9 @@ public class FluxBuilder {
// process stream definitions
buildStreamDefinitions(context, builder);
+ // create worker hooks
+ buildWorkerHooks(context, builder);
+
topology = builder.createTopology();
} else {
// user class supplied...
@@ -478,6 +483,18 @@ public class FluxBuilder {
}
}
+ /**
+ * Given a list of worker hook definitions, build a Storm worker hook implementation by attempting to find a matching
+ * constructor in the given worker hook class and add them to the topology builder.
+ */
+ private static void buildWorkerHooks(ExecutionContext context, TopologyBuilder builder) throws ClassNotFoundException,
+ NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, NoSuchFieldException {
+ for (WorkerHookDef whDef: context.getTopologyDef().getWorkerHooks()) {
+ IWorkerHook workerHook = (IWorkerHook) buildObject(whDef, context);
+ builder.addWorkerHook(workerHook);
+ }
+ }
+
/**
* Given a list of constructor arguments, and a target class, attempt to find a suitable constructor.
*/
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
index a0b4a0329..7744e4c2d 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
@@ -53,6 +53,7 @@ public class TopologyDef {
private Map<String, BoltDef> boltMap = new LinkedHashMap<String, BoltDef>();
private Map<String, SpoutDef> spoutMap = new LinkedHashMap<String, SpoutDef>();
private List<StreamDef> streams = new ArrayList<StreamDef>();
+ private Map<String, WorkerHookDef> workerHookMap = new LinkedHashMap<>();
public String getName() {
@@ -163,6 +164,27 @@ public class TopologyDef {
this.includes = includes;
}
+ /**
+ * Returns worker hook definitions.
+ * @return worker hook definitions
+ */
+ public List<WorkerHookDef> getWorkerHooks() {
+ ArrayList<WorkerHookDef> retval = new ArrayList<>();
+ retval.addAll(this.workerHookMap.values());
+ return retval;
+ }
+
+ /**
+ * Sets worker hook definitions.
+ * @param workerHooks worker hook definitions
+ */
+ public void setWorkerHooks(List<WorkerHookDef> workerHooks) {
+ this.workerHookMap = new LinkedHashMap<>();
+ for (WorkerHookDef workerHook : workerHooks) {
+ this.workerHookMap.put(workerHook.getId(), workerHook);
+ }
+ }
+
// utility methods
public int parallelismForBolt(String boltId) {
return this.boltMap.get(boltId).getParallelism();
@@ -180,6 +202,10 @@ public class TopologyDef {
return this.componentMap.get(id);
}
+ public WorkerHookDef getWorkerHook(String id) {
+ return this.workerHookMap.get(id);
+ }
+
/**
* Adds a list of bolt definitions. Optionally overriding existing definitions
* if one with the same ID already exists.
@@ -243,6 +269,23 @@ public class TopologyDef {
this.streams.addAll(streams);
}
+ /**
+ * Adds a list of worker hook definitions. Optionally overriding existing definitions
+ * if one with the same ID already exists.
+ * @param workerHooks worker hook definitions
+ * @param override whether or not to override existing definitions
+ */
+ public void addAllWorkerHooks(List<WorkerHookDef> workerHooks, boolean override) {
+ for (WorkerHookDef workerHook : workerHooks) {
+ String id = workerHook.getId();
+ if (this.workerHookMap.get(id) == null || override) {
+ this.workerHookMap.put(workerHook.getId(), workerHook);
+ } else {
+ LOG.warn("Ignoring attempt to create worker hook '{}' with override == false.", id);
+ }
+ }
+ }
+
public TopologySourceDef getTopologySource() {
return topologySource;
}
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/WorkerHookDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/WorkerHookDef.java
new file mode 100644
index 000000000..05d717b87
--- /dev/null
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/WorkerHookDef.java
@@ -0,0 +1,25 @@
+/*
+ * 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.flux.model;
+
+/**
+ * Bean representation of a Storm worker hook.
+ */
+public class WorkerHookDef extends BeanDef {
+}
diff --git a/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java b/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
index f5bb25a55..951678573 100644
--- a/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
+++ b/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
@@ -278,4 +278,17 @@ public class TCKTest {
IllegalArgumentException expectedException = assertThrows(IllegalArgumentException.class, () -> FluxBuilder.buildTopology(context));
assertTrue(expectedException.getMessage().contains("Couldn't find a suitable static method"));
}
+
+ @Test
+ public void testTopologyWithWorkerHook() throws Exception {
+ TopologyDef topologyDef = FluxParser.parseResource("/configs/worker_hook.yaml", false, true, null, false);
+ Config conf = FluxBuilder.buildConfig(topologyDef);
+ ExecutionContext context = new ExecutionContext(topologyDef, conf);
+ StormTopology topology = FluxBuilder.buildTopology(context);
+ assertNotNull(topology);
+ assertTrue(topologyDef.getName().equals("worker-hook-topology"));
+ assertTrue(topologyDef.getWorkerHooks().size() > 0);
+ assertTrue(topology.get_worker_hooks_size() > 0);
+ topology.validate();
+ }
}
diff --git a/flux/flux-core/src/test/resources/configs/worker_hook.yaml b/flux/flux-core/src/test/resources/configs/worker_hook.yaml
new file mode 100644
index 000000000..65e54fc70
--- /dev/null
+++ b/flux/flux-core/src/test/resources/configs/worker_hook.yaml
@@ -0,0 +1,47 @@
+# 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.
+---
+name: "worker-hook-topology"
+
+config:
+ topology.workers: 1
+
+# spout definitions
+spouts:
+ - id: "spout-1"
+ className: "org.apache.storm.testing.TestWordSpout"
+ parallelism: 1
+ # ...
+
+# bolt definitions
+bolts:
+ - id: "bolt-1"
+ className: "org.apache.storm.testing.TestWordCounter"
+ parallelism: 1
+
+#stream definitions
+# stream definitions define connections between spouts and bolts.
+# note that such connections can be cyclical
+streams:
+ - name: "spout-1 --> bolt-1" # name isn't used (placeholder for logging, UI, etc.)
+ from: "spout-1"
+ to: "bolt-1"
+ grouping:
+ type: SHUFFLE
+
+workerHooks:
+ - id: "base-worker-hook"
+ className: "org.apache.storm.hooks.BaseWorkerHook"
[storm] 04/04: update README
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 96cc304b0e9acaa2cd0b8bcbb0c95e348334d751
Author: Bharat Gulati <bh...@gmail.com>
AuthorDate: Mon Jul 24 09:55:40 2023 +0530
update README
---
docs/flux.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/flux.md b/docs/flux.md
index 08c08957d..8b7f09d09 100644
--- a/docs/flux.md
+++ b/docs/flux.md
@@ -771,7 +771,7 @@ The `override` property controls how includes affect the values defined in the c
**N.B.:** Includes are not yet recursive. Includes from included files will be ignored.
## Worker Hooks
-Flux allows you to attach topology component that can be executed when a worker starts, and when a worker shuts down. It can be useful when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down, e.g. managing application context. Worker Hooks should be an implementation of [IWorkerHook]({{page.git-blob-base}}/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java). Other than that, they follow similar Bean definition s [...]
+Flux allows you to attach topology components that can be executed when a worker starts, and when a worker shuts down. It can be useful when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down, e.g. managing application context. Worker Hooks should be an implementation of [IWorkerHook]({{page.git-blob-base}}/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java). Other than that, they follow similar Bean definition [...]
Worker Hooks are specified as a map of bean definitions:
[storm] 03/04: minor correction in flux README
Posted by rz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
commit f75ab785326253450b23e96642641ea9d8bd32e6
Author: Bharat Gulati <bh...@gmail.com>
AuthorDate: Sat Jun 17 15:06:36 2023 +0530
minor correction in flux README
---
docs/flux.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/flux.md b/docs/flux.md
index 75a62e133..08c08957d 100644
--- a/docs/flux.md
+++ b/docs/flux.md
@@ -771,9 +771,9 @@ The `override` property controls how includes affect the values defined in the c
**N.B.:** Includes are not yet recursive. Includes from included files will be ignored.
## Worker Hooks
-Flux allows you to attach topology component that can be executed when a worker starts, and when a worker shuts down. It can be useful when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down, e.g. managing application context. Worker Hooks should be an implementation of [IWorkerHook]({{page.git-blob-base}}/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java). Otherwise, follow similar Bean definition semantics as [...]
+Flux allows you to attach topology component that can be executed when a worker starts, and when a worker shuts down. It can be useful when you want to execute operations before topology processing starts, or cleanup operations before your workers shut down, e.g. managing application context. Worker Hooks should be an implementation of [IWorkerHook]({{page.git-blob-base}}/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java). Other than that, they follow similar Bean definition s [...]
-Worker Hooks are specified as a list of maps:
+Worker Hooks are specified as a map of bean definitions:
```yaml
workerHooks: