You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by abhishekagarwal87 <gi...@git.apache.org> on 2016/03/23 12:29:34 UTC
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
GitHub user abhishekagarwal87 opened a pull request:
https://github.com/apache/storm/pull/1249
STORM-1271: Port backtype.storm.daemon.task to java
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/abhishekagarwal87/storm task
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1249.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1249
----
commit 0e61a606042b932908ee8c9ce8b02fe62798c1bf
Author: Abhishek Agarwal <ab...@inmobi.com>
Date: 2016-03-22T20:57:31Z
STORM-1271: Port backtype.storm.daemon.task to java
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57909930
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private BuiltinMetrics builtInMetrics;
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
+ this.stormConf = (Map) executorData.get("storm-conf");
+ this.componentId = (String) executorData.get("component-id");
+ this.streamComponentToGrouper = (Map<String, Map<String, Grouper>>) executorData.get("stream->component->grouper");
+ this.executorStats = (CommonStats) executorData.get("stats");
+ this.builtInMetrics = BuiltinMetricsUtil.mkData((String) executorData.get("type"), this.executorStats);
+ this.workerTopologyContext = (WorkerTopologyContext) executorData.get("worker-context");
+ this.emitSampler = ConfigUtils.mkStatsSampler(stormConf);
+ this.loadMapping = (LoadMapping) workerData.get("load-mapping");
+ this.systemTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("system-topology"));
+ this.userTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("topology"));
+ this.taskObject = mkTaskObject();
+ this.debug = stormConf.containsKey(Config.TOPOLOGY_DEBUG) && (Boolean) stormConf.get(Config.TOPOLOGY_DEBUG);
+ this.addTaskHooks();
+ }
+
+ public List<Integer> getOutgoingTasks(Integer outTaskId, String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting direct: {}; {} {} {} ", outTaskId, componentId, stream, values);
+ }
+ String targetComponent = workerTopologyContext.getComponentId(outTaskId);
+ Map<String, Grouper> componentGrouping = streamComponentToGrouper.get(stream);
+ Grouper grouping = componentGrouping.get(targetComponent);
+ if (null == grouping) {
+ outTaskId = null;
+ }
+ if (grouping != null && grouping != GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot emitDirect to a task expecting a regular grouping");
+ }
+ new EmitInfo(values, stream, taskId, Collections.singletonList(outTaskId)).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ if (null != outTaskId) {
+ executorStats.transferredTuples(stream, 1);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (null != outTaskId) {
+ return Collections.singletonList(outTaskId);
+ }
+ return null;
+ }
+
+ public List<Integer> getOutgoingTasks(String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting: {} {} {}", componentId, stream, values);
+ }
+ List<Integer> outTasks = new ArrayList<>();
+ if (!streamComponentToGrouper.containsKey(stream)) {
+ throw new IllegalArgumentException("Unknown stream ID: " + stream);
+ }
+ if (null != streamComponentToGrouper.get(stream)) {
+ // null value for __system
+ for (Grouper grouper : streamComponentToGrouper.get(stream).values()) {
+ if (grouper == GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot do regular emit to direct stream");
+ }
+ List<Integer> compTasks = grouper.outTasks(taskId, values, loadMapping);
+ outTasks.addAll(compTasks);
+ }
+ }
+ new EmitInfo(values, stream, taskId, outTasks).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ executorStats.transferredTuples(stream, outTasks.size());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return outTasks;
+ }
+
+ public Tuple getTuple(String stream, List values) {
+ return new TupleImpl(systemTopologyContext, values, systemTopologyContext.getThisTaskId(), stream);
+ }
+
+ public Integer getTaskId() {
+ return taskId;
+ }
+
+ public String getComponentId() {
+ return componentId;
+ }
+
+ public TopologyContext getUserContext() throws IOException {
+ return userTopologyContext;
+ }
+
+ public Object getTaskObject() {
+ return taskObject;
+ }
+
+ public BuiltinMetrics getBuiltInMetrics() {
+ return builtInMetrics;
+ }
+
+ private TopologyContext mkTopologyContextBuilder(StormTopology topology) throws IOException {
+ Map conf = (Map) workerData.get("conf");
+ return new TopologyContext(
+ topology,
+ (Map) workerData.get("storm-conf"),
+ (Map<Integer, String>) workerData.get("task->component"),
+ (Map<String, List<Integer>>) workerData.get("component->sorted-tasks"),
+ (Map<String, Map<String, Fields>>) workerData.get("component->stream->fields"),
+ (String) workerData.get("storm-id"),
+ ConfigUtils.supervisorStormResourcesPath(ConfigUtils.supervisorStormDistRoot(conf, (String) workerData.get("storm-id"))),
+ ConfigUtils.workerPidsRoot(conf, (String) workerData.get("worker-id")),
+ taskId,
+ (Integer) workerData.get("port"),
+ (List<Integer>) workerData.get("task-ids"),
+ (Map<String, Object>) workerData.get("default-shared-resources"),
+ (Map<String, Object>) workerData.get("user-shared-resources"),
+ (Map<String, Object>) executorData.get("shared-executor-data"),
+ (Map<Integer, Map<Integer, Map<String, IMetric>>>) executorData.get("interval->task->metric-registry"),
+ (clojure.lang.Atom) executorData.get("open-or-prepare-was-called?")
--- End diff --
That's right.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57444981
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
--- End diff --
I agree, the name is a bit confusing. I was thinking of renaming it to componentObject but that may again be confusing. Any suggestions? I will add the javadoc nonetheless
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57445950
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/GrouperFactory.java ---
@@ -0,0 +1,110 @@
+/**
+ * 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.daemon;
+
+import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+public class GrouperFactory {
+
+ // A no-op grouper
+ public static Grouper DIRECT = new Grouper() {
--- End diff --
final
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57447354
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/GrouperFactory.java ---
@@ -0,0 +1,110 @@
+/**
+ * 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.daemon;
+
+import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+public class GrouperFactory {
+
+ // A no-op grouper
+ public static Grouper DIRECT = new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return null;
+ }
+ };
+
+ public static Grouper mkGrouper(WorkerTopologyContext context, String componentId, String streamId, Fields outFields,
+ Grouping thriftGrouping,
+ List<Integer> unsortedTargetTasks,
+ Map topoConf) {
+ final Random random = new Random();
+ final List<Integer> targetTasks = Ordering.natural().sortedCopy(unsortedTargetTasks);
+ final int numTasks = targetTasks.size();
+ switch (Thrift.groupingType(thriftGrouping)) {
+ case FIELDS:
+ if (Thrift.isGlobalGrouping(thriftGrouping)) {
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List value, LoadMapping load) {
+ if (targetTasks.isEmpty()) {
+ return null;
+ }
+ // It's possible for target to have multiple tasks if it reads multiple sources
+ return Collections.singletonList(targetTasks.get(0));
+ }
+ };
+ }
+ return new Grouper.FieldsGrouper(outFields, thriftGrouping, targetTasks);
+ case SHUFFLE:
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ case ALL:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return targetTasks;
+ }
+ };
+ case LOCAL_OR_SHUFFLE:
+ Set<Integer> sameTasks = Sets.intersection(Sets.newHashSet(targetTasks), Sets.newHashSet(context.getThisWorkerTasks()));
+ if (sameTasks.isEmpty()) {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ } else {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, new ArrayList<>(sameTasks), topoConf);
+ }
+ case NONE:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ int index = random.nextInt(numTasks);
+ return Collections.singletonList(targetTasks.get(index));
+ }
+ };
+ case CUSTOM_OBJECT:
+ CustomStreamGrouping customObject =
+ (CustomStreamGrouping) Thrift.instantiateJavaObject(thriftGrouping.get_custom_object());
+ return new Grouper.CustomGrouper(customObject, context, componentId, streamId, targetTasks);
--- End diff --
If someone created a CustormStreamGrouping that was also LoadAwareCustomStreamGrouping the LoadAware portion would never be called, because we are now using constructors instead of the builder pattern. This really should be.
```
case CUSTOM_OBJECT:
CustomStreamGrouping
customGrouping =
(CustomStreamGrouping) Thrift.instantiateJavaObject(thriftGrouping.get_custom_object());
if (customGrouping instanceof CustomLoadAwareStreamGrouping) {
return new Grouper.LoadAwareCustomGrouper(customSerialized, context, componentId, streamId, targetTasks);
}
return new Grouper.CustomGrouper(customSerialized, context, componentId, streamId, targetTasks);
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57444794
--- Diff: storm-core/src/clj/org/apache/storm/daemon/executor.clj ---
@@ -15,6 +15,7 @@
;; limitations under the License.
(ns org.apache.storm.daemon.executor
(:use [org.apache.storm.daemon common])
+ (:use [clojure.walk])
--- End diff --
https://github.com/apache/storm/pull/1249/files#diff-6e352b8cfa5485b01db59718464c2dd1R324 `stringify-keys`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57444482
--- Diff: storm-core/src/clj/org/apache/storm/daemon/executor.clj ---
@@ -15,6 +15,7 @@
;; limitations under the License.
(ns org.apache.storm.daemon.executor
(:use [org.apache.storm.daemon common])
+ (:use [clojure.walk])
--- End diff --
I don't see where this import is used. Is it needed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57449306
--- Diff: storm-core/src/jvm/org/apache/storm/utils/ConfigUtils.java ---
@@ -135,7 +137,28 @@ public static int samplingRate(Map conf) {
throw new IllegalArgumentException("Illegal topology.stats.sample.rate in conf: " + rate);
}
- // public static mkStatsSampler // depends on Utils.evenSampler() TODO, this is sth we need to do after util
+ public static Callable<Boolean> evenSampler(final int samplingFreq) {
+ final int start = 0;
+ final Random random = new Random();
+ final MutableInt curr = new MutableInt(-1);
+ final MutableInt target = new MutableInt(random.nextInt(samplingFreq));
--- End diff --
The entire reason for the MutableInt was so clojure could ack more like java. It is going to be a lot faster to just use int instead.
```
public static Callable<Boolean> evenSampler(final int samplingFreq) {
final Random random = new Random();
return new Callable<Boolean>() {
private int curr = -1;
private int target = random.nextInt(samplingFreq);
@Override
public Boolean call() throws Exception {
curr++;
if (curr >= samplingFreq) {
curr = 0;
target = random.nextInt(samplingFreq);
}
return curr == target;
}
};
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57385384
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
--- End diff --
It already is. This needs to be upmerged now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57909562
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/GrouperFactory.java ---
@@ -0,0 +1,110 @@
+/**
+ * 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.daemon;
+
+import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+public class GrouperFactory {
+
+ // A no-op grouper
+ public static Grouper DIRECT = new Grouper() {
--- End diff --
will do
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57447253
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/GrouperFactory.java ---
@@ -0,0 +1,110 @@
+/**
+ * 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.daemon;
+
+import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+public class GrouperFactory {
+
+ // A no-op grouper
+ public static Grouper DIRECT = new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return null;
+ }
+ };
+
+ public static Grouper mkGrouper(WorkerTopologyContext context, String componentId, String streamId, Fields outFields,
+ Grouping thriftGrouping,
+ List<Integer> unsortedTargetTasks,
+ Map topoConf) {
+ final Random random = new Random();
+ final List<Integer> targetTasks = Ordering.natural().sortedCopy(unsortedTargetTasks);
+ final int numTasks = targetTasks.size();
+ switch (Thrift.groupingType(thriftGrouping)) {
+ case FIELDS:
+ if (Thrift.isGlobalGrouping(thriftGrouping)) {
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List value, LoadMapping load) {
+ if (targetTasks.isEmpty()) {
+ return null;
+ }
+ // It's possible for target to have multiple tasks if it reads multiple sources
+ return Collections.singletonList(targetTasks.get(0));
+ }
+ };
+ }
+ return new Grouper.FieldsGrouper(outFields, thriftGrouping, targetTasks);
+ case SHUFFLE:
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ case ALL:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return targetTasks;
+ }
+ };
+ case LOCAL_OR_SHUFFLE:
+ Set<Integer> sameTasks = Sets.intersection(Sets.newHashSet(targetTasks), Sets.newHashSet(context.getThisWorkerTasks()));
+ if (sameTasks.isEmpty()) {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ } else {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, new ArrayList<>(sameTasks), topoConf);
+ }
+ case NONE:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ int index = random.nextInt(numTasks);
+ return Collections.singletonList(targetTasks.get(index));
+ }
+ };
+ case CUSTOM_OBJECT:
+ CustomStreamGrouping customObject =
+ (CustomStreamGrouping) Thrift.instantiateJavaObject(thriftGrouping.get_custom_object());
+ return new Grouper.CustomGrouper(customObject, context, componentId, streamId, targetTasks);
+ case CUSTOM_SERIALIZED:
+ CustomStreamGrouping
+ customSerialized =
+ Utils.javaDeserialize(thriftGrouping.get_custom_serialized(), CustomStreamGrouping.class);
+ return new Grouper.CustomGrouper(customSerialized, context, componentId, streamId, targetTasks);
--- End diff --
If someone created a CustormStreamGrouping that was also LoadAwareCustomStreamGrouping the LoadAware portion would never be called, because we are now using constructors instead of the builder pattern. This really should be.
```
case CUSTOM_SERIALIZED:
CustomStreamGrouping
customGrouping =
Utils.javaDeserialize(thriftGrouping.get_custom_serialized(), CustomStreamGrouping.class);
if (customGrouping instanceof CustomLoadAwareStreamGrouping) {
return new Grouper.LoadAwareCustomGrouper(customSerialized, context, componentId, streamId, targetTasks);
}
return new Grouper.CustomGrouper(customSerialized, context, componentId, streamId, targetTasks);
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57449944
--- Diff: storm-core/src/clj/org/apache/storm/daemon/executor.clj ---
@@ -15,6 +15,7 @@
;; limitations under the License.
(ns org.apache.storm.daemon.executor
(:use [org.apache.storm.daemon common])
+ (:use [clojure.walk])
--- End diff --
Ah OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on the pull request:
https://github.com/apache/storm/pull/1249#issuecomment-201179242
up-merged with builtin-metrics changes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57144340
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
--- End diff --
TODO is for the change after PR for porting builtin-metrics is merged.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57447684
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private BuiltinMetrics builtInMetrics;
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
+ this.stormConf = (Map) executorData.get("storm-conf");
+ this.componentId = (String) executorData.get("component-id");
+ this.streamComponentToGrouper = (Map<String, Map<String, Grouper>>) executorData.get("stream->component->grouper");
+ this.executorStats = (CommonStats) executorData.get("stats");
+ this.builtInMetrics = BuiltinMetricsUtil.mkData((String) executorData.get("type"), this.executorStats);
+ this.workerTopologyContext = (WorkerTopologyContext) executorData.get("worker-context");
+ this.emitSampler = ConfigUtils.mkStatsSampler(stormConf);
+ this.loadMapping = (LoadMapping) workerData.get("load-mapping");
+ this.systemTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("system-topology"));
+ this.userTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("topology"));
+ this.taskObject = mkTaskObject();
+ this.debug = stormConf.containsKey(Config.TOPOLOGY_DEBUG) && (Boolean) stormConf.get(Config.TOPOLOGY_DEBUG);
+ this.addTaskHooks();
+ }
+
+ public List<Integer> getOutgoingTasks(Integer outTaskId, String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting direct: {}; {} {} {} ", outTaskId, componentId, stream, values);
+ }
+ String targetComponent = workerTopologyContext.getComponentId(outTaskId);
+ Map<String, Grouper> componentGrouping = streamComponentToGrouper.get(stream);
+ Grouper grouping = componentGrouping.get(targetComponent);
+ if (null == grouping) {
+ outTaskId = null;
+ }
+ if (grouping != null && grouping != GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot emitDirect to a task expecting a regular grouping");
+ }
+ new EmitInfo(values, stream, taskId, Collections.singletonList(outTaskId)).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ if (null != outTaskId) {
+ executorStats.transferredTuples(stream, 1);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (null != outTaskId) {
+ return Collections.singletonList(outTaskId);
+ }
+ return null;
+ }
+
+ public List<Integer> getOutgoingTasks(String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting: {} {} {}", componentId, stream, values);
+ }
+ List<Integer> outTasks = new ArrayList<>();
+ if (!streamComponentToGrouper.containsKey(stream)) {
+ throw new IllegalArgumentException("Unknown stream ID: " + stream);
+ }
+ if (null != streamComponentToGrouper.get(stream)) {
+ // null value for __system
+ for (Grouper grouper : streamComponentToGrouper.get(stream).values()) {
+ if (grouper == GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot do regular emit to direct stream");
+ }
+ List<Integer> compTasks = grouper.outTasks(taskId, values, loadMapping);
+ outTasks.addAll(compTasks);
+ }
+ }
+ new EmitInfo(values, stream, taskId, outTasks).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ executorStats.transferredTuples(stream, outTasks.size());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return outTasks;
+ }
+
+ public Tuple getTuple(String stream, List values) {
+ return new TupleImpl(systemTopologyContext, values, systemTopologyContext.getThisTaskId(), stream);
+ }
+
+ public Integer getTaskId() {
+ return taskId;
+ }
+
+ public String getComponentId() {
+ return componentId;
+ }
+
+ public TopologyContext getUserContext() throws IOException {
+ return userTopologyContext;
+ }
+
+ public Object getTaskObject() {
+ return taskObject;
+ }
+
+ public BuiltinMetrics getBuiltInMetrics() {
+ return builtInMetrics;
+ }
+
+ private TopologyContext mkTopologyContextBuilder(StormTopology topology) throws IOException {
--- End diff --
It looks like `mk-topology-context-builder`, `system-topology-context`, and `user-topology-context` all got merged into this. It probably would be best to just call it `mkTopologyContext`, and eventually have a Constructor on `TopologyContext` that just takes a StormTopology.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57385605
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
--- End diff --
I assume ExecutorData is going to become a java object at some point in the future, and when that happens these will change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1249#issuecomment-203580108
The changes look good but STORM-1663 and STORM-1665 are currently making master unusable so I don't want to merge new things in until they are fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57911963
--- Diff: storm-core/src/jvm/org/apache/storm/utils/ConfigUtils.java ---
@@ -135,7 +137,28 @@ public static int samplingRate(Map conf) {
throw new IllegalArgumentException("Illegal topology.stats.sample.rate in conf: " + rate);
}
- // public static mkStatsSampler // depends on Utils.evenSampler() TODO, this is sth we need to do after util
+ public static Callable<Boolean> evenSampler(final int samplingFreq) {
+ final int start = 0;
+ final Random random = new Random();
+ final MutableInt curr = new MutableInt(-1);
+ final MutableInt target = new MutableInt(random.nextInt(samplingFreq));
--- End diff --
will do. Out of Curiosity, are the calls to emitSampler object done in single thread?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57420085
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
--- End diff --
@revans2 - That is correct. Constructor of Task object remains the same this way (only input types change).
@hustfxj - I guess task should go before executor is ported.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57445169
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Grouper.java ---
@@ -0,0 +1,111 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadAwareShuffleGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.grouping.ShuffleGrouping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.TupleUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public abstract class Grouper {
--- End diff --
Why is this not a `LoadAwareCustomStreamGrouping`? Or perhaps better a different interface that LoadAwareCustomStreamGrouping can extend?
That way we would not need as much glue code to make custom groupings work. It also would make it more consistent in the naming of methods with `chooseTasks` instead of `outTasks`. This is minor, and a bit for my curiosity if you had a reason because adding in more class hierarchy has its drawbacks as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57424616
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
--- End diff --
@knusbaum - Yes. I have removed it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57909402
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Grouper.java ---
@@ -0,0 +1,111 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadAwareShuffleGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.grouping.ShuffleGrouping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.TupleUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public abstract class Grouper {
--- End diff --
Honestly I failed to notice that LoadAwareCustomStreamGrouping has the same interface needed by Task class. It will fit in perfectly. The only caveat is the naming as many implementations of `LoadAwareCustomStreamGrouping` wouldn't exactly be load aware.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1249
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1249#issuecomment-203561593
+1 looks good to me
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57414852
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
--- End diff --
@abhishekagarwal87 You can make ExecutorData and WorkerData become a java objects now, In fact I have already create them when I port the executor.clj to java.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57912312
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
--- End diff --
componentObject is ok for me
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57909644
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/GrouperFactory.java ---
@@ -0,0 +1,110 @@
+/**
+ * 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.daemon;
+
+import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+public class GrouperFactory {
+
+ // A no-op grouper
+ public static Grouper DIRECT = new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return null;
+ }
+ };
+
+ public static Grouper mkGrouper(WorkerTopologyContext context, String componentId, String streamId, Fields outFields,
+ Grouping thriftGrouping,
+ List<Integer> unsortedTargetTasks,
+ Map topoConf) {
+ final Random random = new Random();
+ final List<Integer> targetTasks = Ordering.natural().sortedCopy(unsortedTargetTasks);
+ final int numTasks = targetTasks.size();
+ switch (Thrift.groupingType(thriftGrouping)) {
+ case FIELDS:
+ if (Thrift.isGlobalGrouping(thriftGrouping)) {
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List value, LoadMapping load) {
+ if (targetTasks.isEmpty()) {
+ return null;
+ }
+ // It's possible for target to have multiple tasks if it reads multiple sources
+ return Collections.singletonList(targetTasks.get(0));
+ }
+ };
+ }
+ return new Grouper.FieldsGrouper(outFields, thriftGrouping, targetTasks);
+ case SHUFFLE:
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ case ALL:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return targetTasks;
+ }
+ };
+ case LOCAL_OR_SHUFFLE:
+ Set<Integer> sameTasks = Sets.intersection(Sets.newHashSet(targetTasks), Sets.newHashSet(context.getThisWorkerTasks()));
+ if (sameTasks.isEmpty()) {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, targetTasks, topoConf);
+ } else {
+ return new Grouper.ShuffleGrouper(context, componentId, streamId, new ArrayList<>(sameTasks), topoConf);
+ }
+ case NONE:
+ return new Grouper() {
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ int index = random.nextInt(numTasks);
+ return Collections.singletonList(targetTasks.get(index));
+ }
+ };
+ case CUSTOM_OBJECT:
+ CustomStreamGrouping customObject =
+ (CustomStreamGrouping) Thrift.instantiateJavaObject(thriftGrouping.get_custom_object());
+ return new Grouper.CustomGrouper(customObject, context, componentId, streamId, targetTasks);
--- End diff --
Good catch. This problem will go away once I refactor the classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57909901
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private BuiltinMetrics builtInMetrics;
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
+ this.stormConf = (Map) executorData.get("storm-conf");
+ this.componentId = (String) executorData.get("component-id");
+ this.streamComponentToGrouper = (Map<String, Map<String, Grouper>>) executorData.get("stream->component->grouper");
+ this.executorStats = (CommonStats) executorData.get("stats");
+ this.builtInMetrics = BuiltinMetricsUtil.mkData((String) executorData.get("type"), this.executorStats);
+ this.workerTopologyContext = (WorkerTopologyContext) executorData.get("worker-context");
+ this.emitSampler = ConfigUtils.mkStatsSampler(stormConf);
+ this.loadMapping = (LoadMapping) workerData.get("load-mapping");
+ this.systemTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("system-topology"));
+ this.userTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("topology"));
+ this.taskObject = mkTaskObject();
+ this.debug = stormConf.containsKey(Config.TOPOLOGY_DEBUG) && (Boolean) stormConf.get(Config.TOPOLOGY_DEBUG);
+ this.addTaskHooks();
+ }
+
+ public List<Integer> getOutgoingTasks(Integer outTaskId, String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting direct: {}; {} {} {} ", outTaskId, componentId, stream, values);
+ }
+ String targetComponent = workerTopologyContext.getComponentId(outTaskId);
+ Map<String, Grouper> componentGrouping = streamComponentToGrouper.get(stream);
+ Grouper grouping = componentGrouping.get(targetComponent);
+ if (null == grouping) {
+ outTaskId = null;
+ }
+ if (grouping != null && grouping != GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot emitDirect to a task expecting a regular grouping");
+ }
+ new EmitInfo(values, stream, taskId, Collections.singletonList(outTaskId)).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ if (null != outTaskId) {
+ executorStats.transferredTuples(stream, 1);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (null != outTaskId) {
+ return Collections.singletonList(outTaskId);
+ }
+ return null;
+ }
+
+ public List<Integer> getOutgoingTasks(String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting: {} {} {}", componentId, stream, values);
+ }
+ List<Integer> outTasks = new ArrayList<>();
+ if (!streamComponentToGrouper.containsKey(stream)) {
+ throw new IllegalArgumentException("Unknown stream ID: " + stream);
+ }
+ if (null != streamComponentToGrouper.get(stream)) {
+ // null value for __system
+ for (Grouper grouper : streamComponentToGrouper.get(stream).values()) {
+ if (grouper == GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot do regular emit to direct stream");
+ }
+ List<Integer> compTasks = grouper.outTasks(taskId, values, loadMapping);
+ outTasks.addAll(compTasks);
+ }
+ }
+ new EmitInfo(values, stream, taskId, outTasks).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ executorStats.transferredTuples(stream, outTasks.size());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return outTasks;
+ }
+
+ public Tuple getTuple(String stream, List values) {
+ return new TupleImpl(systemTopologyContext, values, systemTopologyContext.getThisTaskId(), stream);
+ }
+
+ public Integer getTaskId() {
+ return taskId;
+ }
+
+ public String getComponentId() {
+ return componentId;
+ }
+
+ public TopologyContext getUserContext() throws IOException {
+ return userTopologyContext;
+ }
+
+ public Object getTaskObject() {
+ return taskObject;
+ }
+
+ public BuiltinMetrics getBuiltInMetrics() {
+ return builtInMetrics;
+ }
+
+ private TopologyContext mkTopologyContextBuilder(StormTopology topology) throws IOException {
--- End diff --
This makes sense. I will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57447877
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private BuiltinMetrics builtInMetrics;
+ private boolean debug;
+
+ public Task(Map executorData, Integer taskId) throws IOException {
+ this.taskId = taskId;
+ this.executorData = executorData;
+ this.workerData = (Map) executorData.get("worker");
+ this.stormConf = (Map) executorData.get("storm-conf");
+ this.componentId = (String) executorData.get("component-id");
+ this.streamComponentToGrouper = (Map<String, Map<String, Grouper>>) executorData.get("stream->component->grouper");
+ this.executorStats = (CommonStats) executorData.get("stats");
+ this.builtInMetrics = BuiltinMetricsUtil.mkData((String) executorData.get("type"), this.executorStats);
+ this.workerTopologyContext = (WorkerTopologyContext) executorData.get("worker-context");
+ this.emitSampler = ConfigUtils.mkStatsSampler(stormConf);
+ this.loadMapping = (LoadMapping) workerData.get("load-mapping");
+ this.systemTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("system-topology"));
+ this.userTopologyContext = mkTopologyContextBuilder((StormTopology) workerData.get("topology"));
+ this.taskObject = mkTaskObject();
+ this.debug = stormConf.containsKey(Config.TOPOLOGY_DEBUG) && (Boolean) stormConf.get(Config.TOPOLOGY_DEBUG);
+ this.addTaskHooks();
+ }
+
+ public List<Integer> getOutgoingTasks(Integer outTaskId, String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting direct: {}; {} {} {} ", outTaskId, componentId, stream, values);
+ }
+ String targetComponent = workerTopologyContext.getComponentId(outTaskId);
+ Map<String, Grouper> componentGrouping = streamComponentToGrouper.get(stream);
+ Grouper grouping = componentGrouping.get(targetComponent);
+ if (null == grouping) {
+ outTaskId = null;
+ }
+ if (grouping != null && grouping != GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot emitDirect to a task expecting a regular grouping");
+ }
+ new EmitInfo(values, stream, taskId, Collections.singletonList(outTaskId)).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ if (null != outTaskId) {
+ executorStats.transferredTuples(stream, 1);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (null != outTaskId) {
+ return Collections.singletonList(outTaskId);
+ }
+ return null;
+ }
+
+ public List<Integer> getOutgoingTasks(String stream, List<Object> values) {
+ if (debug) {
+ LOG.info("Emitting: {} {} {}", componentId, stream, values);
+ }
+ List<Integer> outTasks = new ArrayList<>();
+ if (!streamComponentToGrouper.containsKey(stream)) {
+ throw new IllegalArgumentException("Unknown stream ID: " + stream);
+ }
+ if (null != streamComponentToGrouper.get(stream)) {
+ // null value for __system
+ for (Grouper grouper : streamComponentToGrouper.get(stream).values()) {
+ if (grouper == GrouperFactory.DIRECT) {
+ throw new IllegalArgumentException("Cannot do regular emit to direct stream");
+ }
+ List<Integer> compTasks = grouper.outTasks(taskId, values, loadMapping);
+ outTasks.addAll(compTasks);
+ }
+ }
+ new EmitInfo(values, stream, taskId, outTasks).applyOn(userTopologyContext);
+ try {
+ if (emitSampler.call()) {
+ executorStats.emittedTuple(stream);
+ executorStats.transferredTuples(stream, outTasks.size());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return outTasks;
+ }
+
+ public Tuple getTuple(String stream, List values) {
+ return new TupleImpl(systemTopologyContext, values, systemTopologyContext.getThisTaskId(), stream);
+ }
+
+ public Integer getTaskId() {
+ return taskId;
+ }
+
+ public String getComponentId() {
+ return componentId;
+ }
+
+ public TopologyContext getUserContext() throws IOException {
+ return userTopologyContext;
+ }
+
+ public Object getTaskObject() {
+ return taskObject;
+ }
+
+ public BuiltinMetrics getBuiltInMetrics() {
+ return builtInMetrics;
+ }
+
+ private TopologyContext mkTopologyContextBuilder(StormTopology topology) throws IOException {
+ Map conf = (Map) workerData.get("conf");
+ return new TopologyContext(
+ topology,
+ (Map) workerData.get("storm-conf"),
+ (Map<Integer, String>) workerData.get("task->component"),
+ (Map<String, List<Integer>>) workerData.get("component->sorted-tasks"),
+ (Map<String, Map<String, Fields>>) workerData.get("component->stream->fields"),
+ (String) workerData.get("storm-id"),
+ ConfigUtils.supervisorStormResourcesPath(ConfigUtils.supervisorStormDistRoot(conf, (String) workerData.get("storm-id"))),
+ ConfigUtils.workerPidsRoot(conf, (String) workerData.get("worker-id")),
+ taskId,
+ (Integer) workerData.get("port"),
+ (List<Integer>) workerData.get("task-ids"),
+ (Map<String, Object>) workerData.get("default-shared-resources"),
+ (Map<String, Object>) workerData.get("user-shared-resources"),
+ (Map<String, Object>) executorData.get("shared-executor-data"),
+ (Map<Integer, Map<Integer, Map<String, IMetric>>>) executorData.get("interval->task->metric-registry"),
+ (clojure.lang.Atom) executorData.get("open-or-prepare-was-called?")
--- End diff --
I assume this will go to AtomicBoolean when Executor goes to java.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57380470
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,247 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
+ private Map stormConf;
+ private Callable<Boolean> emitSampler;
+ private CommonStats executorStats;
+ private Map<String, Map<String, Grouper>> streamComponentToGrouper;
+ private Object builtInMetrics; //TODO
--- End diff --
I don't think we want any TODO comments in the code. I'm guessing builtin-metrics should be merged before this?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1249#issuecomment-201319899
Overall it looks really good. Only one real bug that I could see, but then some nits and design choice questions, mostly around if what we did for clojure still makes since in java, all of the design stuff I see as very minor and we can address it later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57445892
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Grouper.java ---
@@ -0,0 +1,111 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.grouping.CustomStreamGrouping;
+import org.apache.storm.grouping.LoadAwareShuffleGrouping;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.grouping.ShuffleGrouping;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.TupleUtils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public abstract class Grouper {
+
+ public abstract List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load);
+
+ public static class FieldsGrouper extends Grouper {
+
+ private final Fields outFields;
+ private final List<Integer> targetTasks;
+ private final Fields groupFields;
+ private final int numTasks;
+
+ public FieldsGrouper(Fields outFields, Grouping thriftGrouping,
+ List<Integer> targetTasks) {
+ this.outFields = outFields;
+ this.targetTasks = targetTasks;
+ this.groupFields = new Fields(Thrift.fieldGrouping(thriftGrouping));
+ this.numTasks = targetTasks.size();
+ }
+
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ int targetTaskIndex = Math.abs(TupleUtils.listHashCode(outFields.select(groupFields, values))) % numTasks;
+ return Collections.singletonList(targetTasks.get(targetTaskIndex));
+ }
+ }
+
+ public static class ShuffleGrouper extends Grouper {
+
+ private Grouper delegate;
+
+ public ShuffleGrouper(WorkerTopologyContext context, String componentId, String streamId,
+ List<Integer> targetTasks, Map topoConf) {
+ if (null != topoConf.get(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING) && (boolean) topoConf
+ .get(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING)) {
+ delegate = new CustomGrouper(new ShuffleGrouping(), context, componentId, streamId, targetTasks);
+ } else {
+ delegate = new LoadAwareCustomGrouper(new LoadAwareShuffleGrouping(), context, componentId, streamId, targetTasks);
+ }
+ }
+
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return delegate.outTasks(taskId, values, load);
+ }
+ }
+
+ public static class CustomGrouper extends Grouper {
+
+ protected final CustomStreamGrouping customStreamGrouping;
+
+ public CustomGrouper(CustomStreamGrouping customStreamGrouping, WorkerTopologyContext context, String componentId, String streamId,
+ List<Integer> targetTasks) {
+ this.customStreamGrouping = customStreamGrouping;
+ this.customStreamGrouping.prepare(context, new GlobalStreamId(componentId, streamId), targetTasks);
+ }
+
+ @Override
+ public List<Integer> outTasks(Integer taskId, List<Object> values, LoadMapping load) {
+ return customStreamGrouping.chooseTasks(taskId, values);
+ }
+ }
+
+ public static class LoadAwareCustomGrouper extends CustomGrouper {
+
+ public LoadAwareCustomGrouper(LoadAwareShuffleGrouping customStreamGrouping, WorkerTopologyContext context, String componentId,
--- End diff --
`LoadAwareShuffleGrouping` should be `LoadAwareCustomStreamGrouping` instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1249#discussion_r57444330
--- Diff: storm-core/src/jvm/org/apache/storm/daemon/Task.java ---
@@ -0,0 +1,246 @@
+/**
+ * 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.daemon;
+
+import org.apache.storm.Config;
+import org.apache.storm.Thrift;
+import org.apache.storm.daemon.metrics.BuiltinMetrics;
+import org.apache.storm.daemon.metrics.BuiltinMetricsUtil;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.ComponentObject;
+import org.apache.storm.generated.JavaObject;
+import org.apache.storm.generated.ShellComponent;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StateSpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.LoadMapping;
+import org.apache.storm.hooks.ITaskHook;
+import org.apache.storm.hooks.info.EmitInfo;
+import org.apache.storm.metric.api.IMetric;
+import org.apache.storm.spout.ShellSpout;
+import org.apache.storm.stats.CommonStats;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class Task {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Task.class);
+
+ private Map executorData;
+ private Map workerData;
+ private TopologyContext systemTopologyContext;
+ private TopologyContext userTopologyContext;
+ private WorkerTopologyContext workerTopologyContext;
+ private LoadMapping loadMapping;
+ private Integer taskId;
+ private String componentId;
+ private Object taskObject;
--- End diff --
It might be nice to have some javadocs explaining that this is the actual bolt/spout instance that is being executed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: STORM-1271: Port backtype.storm.daemon.task to...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on the pull request:
https://github.com/apache/storm/pull/1249#issuecomment-203508780
@revans2 - Addressed your comments. Please review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---