You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by lo...@apache.org on 2016/03/01 00:58:25 UTC
[02/13] storm git commit: [Storm 1245] port
backtype.storm.daemon.acker to java
[Storm 1245] port backtype.storm.daemon.acker to java
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/675b0c4f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/675b0c4f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/675b0c4f
Branch: refs/heads/master
Commit: 675b0c4f786838a13122b6743ca6c946aa1d63ee
Parents: e9dc271
Author: basti.lj <ba...@alibaba-inc.com>
Authored: Wed Feb 3 15:46:08 2016 +0800
Committer: basti.lj <ba...@alibaba-inc.com>
Committed: Wed Feb 3 15:46:08 2016 +0800
----------------------------------------------------------------------
.../src/clj/org/apache/storm/daemon/acker.clj | 57 ++++++++
.../src/clj/org/apache/storm/daemon/common.clj | 16 +--
storm-core/src/clj/org/apache/storm/testing.clj | 11 +-
.../src/jvm/org/apache/storm/daemon/Acker.java | 133 -------------------
.../jvm/org/apache/storm/daemon/AckerBolt.java | 129 ++++++++++++++++++
5 files changed, 197 insertions(+), 149 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/675b0c4f/storm-core/src/clj/org/apache/storm/daemon/acker.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/acker.clj b/storm-core/src/clj/org/apache/storm/daemon/acker.clj
new file mode 100644
index 0000000..9902b35
--- /dev/null
+++ b/storm-core/src/clj/org/apache/storm/daemon/acker.clj
@@ -0,0 +1,57 @@
+;; 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.
+(ns org.apache.storm.daemon.acker
+ (:import [org.apache.storm.task OutputCollector TopologyContext IBolt])
+ (:import [org.apache.storm.tuple Tuple Fields])
+ (:import [org.apache.storm.utils RotatingMap MutableObject])
+ (:import [java.util List Map])
+ (:import [org.apache.storm Constants]
+ (org.apache.storm.daemon AckerBolt))
+ (:use [org.apache.storm config util log])
+ (:gen-class
+ :init init
+ :implements [org.apache.storm.task.IBolt]
+ :constructors {[] []}
+ :state state))
+
+(def ACKER-COMPONENT-ID AckerBolt/ACKER_COMPONENT_ID)
+(def ACKER-INIT-STREAM-ID AckerBolt/ACKER_INIT_STREAM_ID)
+(def ACKER-ACK-STREAM-ID AckerBolt/ACKER_ACK_STREAM_ID)
+(def ACKER-FAIL-STREAM-ID AckerBolt/ACKER_FAIL_STREAM_ID)
+
+(defn mk-acker-bolt []
+ (let [output-collector (MutableObject.)
+ pending (MutableObject.)]
+ (log-message "Symbol AckerBolt" (symbol "AckerBolt") )
+ (AckerBolt.)))
+
+(defn -init []
+ [[] (container)])
+
+(defn -prepare [this conf context collector]
+ (let [^IBolt ret (mk-acker-bolt)]
+ (container-set! (.state ^org.apache.storm.daemon.acker this) ret)
+ (.prepare ret conf context collector)))
+
+(defn -execute [this tuple]
+ (let [^IBolt delegate (container-get (.state ^org.apache.storm.daemon.acker this))]
+ (.execute delegate tuple)
+ ))
+
+(defn -cleanup [this]
+ (let [^IBolt delegate (container-get (.state ^org.apache.storm.daemon.acker this))]
+ (.cleanup delegate)
+ ))
http://git-wip-us.apache.org/repos/asf/storm/blob/675b0c4f/storm-core/src/clj/org/apache/storm/daemon/common.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/daemon/common.clj b/storm-core/src/clj/org/apache/storm/daemon/common.clj
index 45e0582..6ecc918 100644
--- a/storm-core/src/clj/org/apache/storm/daemon/common.clj
+++ b/storm-core/src/clj/org/apache/storm/daemon/common.clj
@@ -15,7 +15,6 @@
;; limitations under the License.
(ns org.apache.storm.daemon.common
(:use [org.apache.storm log config util])
- (:import [org.apache.storm.daemon Acker])
(:import [org.apache.storm.generated StormTopology
InvalidTopologyException GlobalStreamId]
[org.apache.storm.utils ThriftTopologyUtils])
@@ -24,19 +23,20 @@
(:import [org.apache.storm Constants])
(:import [org.apache.storm.metric SystemBolt])
(:import [org.apache.storm.metric EventLoggerBolt])
- (:import [org.apache.storm.security.auth IAuthorizer])
+ (:import [org.apache.storm.security.auth IAuthorizer])
(:import [java.io InterruptedIOException])
- (:require [clojure.set :as set])
+ (:require [clojure.set :as set])
+ (:require [org.apache.storm.daemon.acker :as acker])
(:require [org.apache.storm.thrift :as thrift])
(:require [metrics.reporters.jmx :as jmx]))
(defn start-metrics-reporters []
(jmx/start (jmx/reporter {})))
-(def ACKER-COMPONENT-ID Acker/ACKER_COMPONENT_ID)
-(def ACKER-INIT-STREAM-ID Acker/ACKER_INIT_STREAM_ID)
-(def ACKER-ACK-STREAM-ID Acker/ACKER_ACK_STREAM_ID)
-(def ACKER-FAIL-STREAM-ID Acker/ACKER_FAIL_STREAM_ID)
+(def ACKER-COMPONENT-ID acker/ACKER-COMPONENT-ID)
+(def ACKER-INIT-STREAM-ID acker/ACKER-INIT-STREAM-ID)
+(def ACKER-ACK-STREAM-ID acker/ACKER-ACK-STREAM-ID)
+(def ACKER-FAIL-STREAM-ID acker/ACKER-FAIL-STREAM-ID)
(def SYSTEM-STREAM-ID "__system")
@@ -207,7 +207,7 @@
(defn add-acker! [storm-conf ^StormTopology ret]
(let [num-executors (if (nil? (storm-conf TOPOLOGY-ACKER-EXECUTORS)) (storm-conf TOPOLOGY-WORKERS) (storm-conf TOPOLOGY-ACKER-EXECUTORS))
acker-bolt (thrift/mk-bolt-spec* (acker-inputs ret)
- (Acker. )
+ (new org.apache.storm.daemon.acker)
{ACKER-ACK-STREAM-ID (thrift/direct-output-fields ["id"])
ACKER-FAIL-STREAM-ID (thrift/direct-output-fields ["id"])
}
http://git-wip-us.apache.org/repos/asf/storm/blob/675b0c4f/storm-core/src/clj/org/apache/storm/testing.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/testing.clj b/storm-core/src/clj/org/apache/storm/testing.clj
index 08662ff..cc78659 100644
--- a/storm-core/src/clj/org/apache/storm/testing.clj
+++ b/storm-core/src/clj/org/apache/storm/testing.clj
@@ -45,9 +45,9 @@
(:import [org.apache.storm.tuple Tuple])
(:import [org.apache.storm.generated StormTopology])
(:import [org.apache.storm.task TopologyContext])
- (:import [org.apache.storm.daemon Acker])
(:require [org.apache.storm [zookeeper :as zk]])
(:require [org.apache.storm.messaging.loader :as msg-loader])
+ (:require [org.apache.storm.daemon.acker :as acker])
(:use [org.apache.storm cluster util thrift config log local-state]))
(defn feeder-spout
@@ -612,11 +612,6 @@
(get key)
.get))
-;; Temporary solution. It should be removed after migration.
-(defn mk-acker-bolt
- []
- (Acker.))
-
(defmacro with-tracked-cluster
[[cluster-sym & cluster-args] & body]
`(let [id# (uuid)]
@@ -627,8 +622,8 @@
(.put "transferred" (AtomicInteger. 0))
(.put "processed" (AtomicInteger. 0))))
(with-var-roots
- [mk-acker-bolt
- (let [old# mk-acker-bolt]
+ [acker/mk-acker-bolt
+ (let [old# acker/mk-acker-bolt]
(fn [& args#] (NonRichBoltTracker. (apply old# args#) id#)))
;; critical that this particular function is overridden here,
;; since the transferred stat needs to be incremented at the moment
http://git-wip-us.apache.org/repos/asf/storm/blob/675b0c4f/storm-core/src/jvm/org/apache/storm/daemon/Acker.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/Acker.java b/storm-core/src/jvm/org/apache/storm/daemon/Acker.java
deleted file mode 100644
index 1e38fd7..0000000
--- a/storm-core/src/jvm/org/apache/storm/daemon/Acker.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/**
- * 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.task.IBolt;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.utils.RotatingMap;
-import org.apache.storm.utils.TupleUtils;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-
-public class Acker implements IBolt {
- private static final Logger LOG = LoggerFactory.getLogger(Acker.class);
-
- private static final long serialVersionUID = 4430906880683183091L;
-
- public static final String ACKER_COMPONENT_ID = "__acker";
- public static final String ACKER_INIT_STREAM_ID = "__ack_init";
- public static final String ACKER_ACK_STREAM_ID = "__ack_ack";
- public static final String ACKER_FAIL_STREAM_ID = "__ack_fail";
-
- public static final int TIMEOUT_BUCKET_NUM = 3;
-
- private OutputCollector collector;
- private RotatingMap<Object, AckObject> pending;
-
- private class AckObject {
- public long val = 0L;
- public Integer spoutTask = null;
- public boolean failed = false;
-
- // val xor value
- public void updateAck(Object value) {
- val = Utils.bitXor(val, value);
- }
- }
-
- public Acker() {
-
- }
-
- @Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- this.collector = collector;
- this.pending = new RotatingMap<Object, AckObject>(TIMEOUT_BUCKET_NUM);
- }
-
- @Override
- public void execute(Tuple input) {
- if (TupleUtils.isTick(input)) {
- Map<Object, AckObject> tmp = pending.rotate();
- LOG.debug("Number of timeout tuples:{}", tmp.size());
- }
-
- String streamId = input.getSourceStreamId();
- Object id = input.getValue(0);
- AckObject curr = pending.get(id);
- if (ACKER_INIT_STREAM_ID.equals(streamId)) {
- if (curr == null) {
- curr = new AckObject();
- curr.val = input.getLong(1);
- curr.spoutTask = input.getInteger(2);
- pending.put(id, curr);
- } else {
- // If receiving bolt's ack before the init message from spout, just update the xor value.
- curr.updateAck(input.getValue(1));
- curr.spoutTask = input.getInteger(2);
- }
- } else if (ACKER_ACK_STREAM_ID.equals(streamId)) {
- if (curr != null) {
- curr.updateAck(input.getValue(1));
- } else {
- curr = new AckObject();
- curr.val = input.getLong(1);
- pending.put(id, curr);
- }
- } else if (ACKER_FAIL_STREAM_ID.equals(streamId)) {
- if (curr == null) {
- // The tuple has been already timeout or failed. So, do nothing
- return;
- }
- curr.failed = true;
- } else {
- LOG.warn("Unknown source stream {} from task-{}", streamId, input.getSourceTask());
- return;
- }
-
- Integer task = curr.spoutTask;
- if (task != null) {
- if (curr.val == 0) {
- pending.remove(id);
- List values = Utils.mkList(id);
- collector.emitDirect(task, ACKER_ACK_STREAM_ID, values);
- } else {
- if (curr.failed) {
- pending.remove(id);
- List values = Utils.mkList(id);
- collector.emitDirect(task, ACKER_FAIL_STREAM_ID, values);
- }
- }
- } else {
-
- }
-
- collector.ack(input);
- }
-
- @Override
- public void cleanup() {
-
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/675b0c4f/storm-core/src/jvm/org/apache/storm/daemon/AckerBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/AckerBolt.java b/storm-core/src/jvm/org/apache/storm/daemon/AckerBolt.java
new file mode 100644
index 0000000..80ed4ca
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/daemon/AckerBolt.java
@@ -0,0 +1,129 @@
+/**
+ * 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.task.IBolt;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.RotatingMap;
+import org.apache.storm.utils.TupleUtils;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class AckerBolt implements IBolt {
+ private static final Logger LOG = LoggerFactory.getLogger(AckerBolt.class);
+
+ private static final long serialVersionUID = 4430906880683183091L;
+
+ public static final String ACKER_COMPONENT_ID = "__acker";
+ public static final String ACKER_INIT_STREAM_ID = "__ack_init";
+ public static final String ACKER_ACK_STREAM_ID = "__ack_ack";
+ public static final String ACKER_FAIL_STREAM_ID = "__ack_fail";
+
+ public static final int TIMEOUT_BUCKET_NUM = 3;
+
+ private OutputCollector collector;
+ private RotatingMap<Object, AckObject> pending;
+
+ private class AckObject {
+ public long val = 0L;
+ public Integer spoutTask = null;
+ public boolean failed = false;
+
+ // val xor value
+ public void updateAck(Object value) {
+ val = Utils.bitXor(val, value);
+ }
+ }
+
+ @Override
+ public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ this.collector = collector;
+ this.pending = new RotatingMap<Object, AckObject>(TIMEOUT_BUCKET_NUM);
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ if (TupleUtils.isTick(input)) {
+ Map<Object, AckObject> tmp = pending.rotate();
+ LOG.debug("Number of timeout tuples:{}", tmp.size());
+ }
+
+ String streamId = input.getSourceStreamId();
+ Object id = input.getValue(0);
+ AckObject curr = pending.get(id);
+ if (ACKER_INIT_STREAM_ID.equals(streamId)) {
+ if (curr == null) {
+ curr = new AckObject();
+ curr.val = input.getLong(1);
+ curr.spoutTask = input.getInteger(2);
+ pending.put(id, curr);
+ } else {
+ // If receiving bolt's ack before the init message from spout, just update the xor value.
+ curr.updateAck(input.getValue(1));
+ curr.spoutTask = input.getInteger(2);
+ }
+ } else if (ACKER_ACK_STREAM_ID.equals(streamId)) {
+ if (curr != null) {
+ curr.updateAck(input.getValue(1));
+ } else {
+ curr = new AckObject();
+ curr.val = input.getLong(1);
+ pending.put(id, curr);
+ }
+ } else if (ACKER_FAIL_STREAM_ID.equals(streamId)) {
+ if (curr == null) {
+ // The tuple has been already timeout or failed. So, do nothing
+ return;
+ }
+ curr.failed = true;
+ } else {
+ LOG.warn("Unknown source stream {} from task-{}", streamId, input.getSourceTask());
+ return;
+ }
+
+ Integer task = curr.spoutTask;
+ if (task != null) {
+ if (curr.val == 0) {
+ pending.remove(id);
+ List values = Utils.mkList(id);
+ collector.emitDirect(task, ACKER_ACK_STREAM_ID, values);
+ } else {
+ if (curr.failed) {
+ pending.remove(id);
+ List values = Utils.mkList(id);
+ collector.emitDirect(task, ACKER_FAIL_STREAM_ID, values);
+ }
+ }
+ } else {
+
+ }
+
+ collector.ack(input);
+ }
+
+ @Override
+ public void cleanup() {
+
+ }
+}
\ No newline at end of file