You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2016/10/11 16:18:04 UTC
[1/2] storm git commit: Merge branch
'Apache_master_STORM-2097_TridentLogs' of
https://github.com/hmcl/storm-apache into STORM-2097
Repository: storm
Updated Branches:
refs/heads/1.x-branch 555146fc6 -> d7d7e61a1
Merge branch 'Apache_master_STORM-2097_TridentLogs' of https://github.com/hmcl/storm-apache into STORM-2097
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/91903743
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/91903743
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/91903743
Branch: refs/heads/1.x-branch
Commit: 91903743d3c3acb2ac5e4949629f2e1b52cc1ed9
Parents: 555146f
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Tue Oct 11 08:45:35 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Tue Oct 11 09:06:29 2016 -0700
----------------------------------------------------------------------
.../starter/spout/RandomSentenceSpout.java | 39 ++++++++++-
.../starter/trident/DebugMemoryMapState.java | 73 ++++++++++++++++++++
.../starter/trident/TridentKafkaWordCount.java | 58 +++++++++++-----
.../jvm/org/apache/storm/kafka/KafkaUtils.java | 9 ++-
.../trident/TransactionalTridentKafkaSpout.java | 8 +--
.../kafka/trident/TridentKafkaEmitter.java | 30 ++++++--
.../storm/trident/operation/builtin/Debug.java | 18 ++++-
.../spout/IOpaquePartitionedTridentSpout.java | 11 +--
.../trident/spout/IPartitionedTridentSpout.java | 10 +--
.../OpaquePartitionedTridentSpoutExecutor.java | 62 +++++++++++++----
.../spout/PartitionedTridentSpoutExecutor.java | 24 +++++--
.../trident/state/CombinerValueUpdater.java | 9 +++
.../topology/MasterBatchCoordinator.java | 39 +++++++++--
.../state/RotatingTransactionalState.java | 59 +++++++++++++---
.../topology/state/TransactionalState.java | 44 ++++++++----
15 files changed, 399 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/examples/storm-starter/src/jvm/org/apache/storm/starter/spout/RandomSentenceSpout.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/spout/RandomSentenceSpout.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/spout/RandomSentenceSpout.java
index 49bec2e..0fe28e2 100644
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/spout/RandomSentenceSpout.java
+++ b/examples/storm-starter/src/jvm/org/apache/storm/starter/spout/RandomSentenceSpout.java
@@ -24,11 +24,17 @@ import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.text.SimpleDateFormat;
+import java.util.Date;
import java.util.Map;
import java.util.Random;
public class RandomSentenceSpout extends BaseRichSpout {
+ private static final Logger LOG = LoggerFactory.getLogger(RandomSentenceSpout.class);
+
SpoutOutputCollector _collector;
Random _rand;
@@ -42,12 +48,19 @@ public class RandomSentenceSpout extends BaseRichSpout {
@Override
public void nextTuple() {
Utils.sleep(100);
- String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away",
- "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" };
- String sentence = sentences[_rand.nextInt(sentences.length)];
+ String[] sentences = new String[]{sentence("the cow jumped over the moon"), sentence("an apple a day keeps the doctor away"),
+ sentence("four score and seven years ago"), sentence("snow white and the seven dwarfs"), sentence("i am at two with nature")};
+ final String sentence = sentences[_rand.nextInt(sentences.length)];
+
+ LOG.debug("Emitting tuple: {}", sentence);
+
_collector.emit(new Values(sentence));
}
+ protected String sentence(String input) {
+ return input;
+ }
+
@Override
public void ack(Object id) {
}
@@ -61,4 +74,24 @@ public class RandomSentenceSpout extends BaseRichSpout {
declarer.declare(new Fields("word"));
}
+ // Add unique identifier to each tuple, which is helpful for debugging
+ public static class TimeStamped extends RandomSentenceSpout {
+ private final String prefix;
+
+ public TimeStamped() {
+ this("");
+ }
+
+ public TimeStamped(String prefix) {
+ this.prefix = prefix;
+ }
+
+ protected String sentence(String input) {
+ return prefix + currentDate() + " " + input;
+ }
+
+ private String currentDate() {
+ return new SimpleDateFormat("yyyy.MM.dd_HH:mm:ss.SSSSSSSSS").format(new Date());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/DebugMemoryMapState.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/DebugMemoryMapState.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/DebugMemoryMapState.java
new file mode 100644
index 0000000..3870afa
--- /dev/null
+++ b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/DebugMemoryMapState.java
@@ -0,0 +1,73 @@
+/*
+ * 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.starter.trident;
+
+import org.apache.storm.task.IMetricsContext;
+import org.apache.storm.topology.FailedException;
+import org.apache.storm.trident.state.CombinerValueUpdater;
+import org.apache.storm.trident.state.State;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.ValueUpdater;
+import org.apache.storm.trident.testing.MemoryMapState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class DebugMemoryMapState<T> extends MemoryMapState<T> {
+ private static final Logger LOG = LoggerFactory.getLogger(DebugMemoryMapState.class);
+
+ private int updateCount = 0;
+
+ public DebugMemoryMapState(String id) {
+ super(id);
+ }
+
+ public List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters) {
+ print(keys, updaters);
+ if ((updateCount++ % 5) == 0) {
+ LOG.error("Throwing FailedException");
+ throw new FailedException("Enforced State Update Fail. On retrial should replay the exact same batch.");
+ }
+ return super.multiUpdate(keys, updaters);
+ }
+
+ private void print(List<List<Object>> keys, List<ValueUpdater> updaters) {
+ for (int i = 0; i < keys.size(); i++) {
+ ValueUpdater valueUpdater = updaters.get(i);
+ Object arg = ((CombinerValueUpdater) valueUpdater).getArg();
+ LOG.debug("updateCount = {}, keys = {} => updaterArgs = {}", updateCount, keys.get(i), arg);
+ }
+ }
+
+ public static class Factory implements StateFactory {
+ String _id;
+
+ public Factory() {
+ _id = UUID.randomUUID().toString();
+ }
+
+ @Override
+ public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
+ return new DebugMemoryMapState(_id + partitionIndex);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
index 0499041..ee8d001 100644
--- a/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
+++ b/examples/storm-starter/src/jvm/org/apache/storm/starter/trident/TridentKafkaWordCount.java
@@ -23,32 +23,37 @@
package org.apache.storm.starter.trident;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
-import org.apache.storm.spout.SchemeAsMultiScheme;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.tuple.Fields;
-import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.storm.kafka.StringScheme;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.kafka.bolt.KafkaBolt;
import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
+import org.apache.storm.kafka.trident.OpaqueTridentKafkaSpout;
import org.apache.storm.kafka.trident.TransactionalTridentKafkaSpout;
import org.apache.storm.kafka.trident.TridentKafkaConfig;
+import org.apache.storm.spout.SchemeAsMultiScheme;
import org.apache.storm.starter.spout.RandomSentenceSpout;
+import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
+import org.apache.storm.trident.operation.builtin.Debug;
import org.apache.storm.trident.operation.builtin.FilterNull;
import org.apache.storm.trident.operation.builtin.MapGet;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.testing.Split;
+import org.apache.storm.tuple.Fields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.Serializable;
import java.util.Properties;
/**
@@ -72,7 +77,8 @@ import java.util.Properties;
* <a href="https://github.com/apache/storm/tree/master/external/storm-kafka"> Storm Kafka </a>.
* </p>
*/
-public class TridentKafkaWordCount {
+public class TridentKafkaWordCount implements Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaWordCount.class);
private String zkUrl;
private String brokerUrl;
@@ -91,7 +97,7 @@ public class TridentKafkaWordCount {
*
* @return a transactional trident kafka spout.
*/
- private TransactionalTridentKafkaSpout createKafkaSpout() {
+ private TransactionalTridentKafkaSpout createTransactionalKafkaSpout() {
ZkHosts hosts = new ZkHosts(zkUrl);
TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
config.scheme = new SchemeAsMultiScheme(new StringScheme());
@@ -101,6 +107,16 @@ public class TridentKafkaWordCount {
return new TransactionalTridentKafkaSpout(config);
}
+ private OpaqueTridentKafkaSpout createOpaqueKafkaSpout() {
+ ZkHosts hosts = new ZkHosts(zkUrl);
+ TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");
+ config.scheme = new SchemeAsMultiScheme(new StringScheme());
+
+ // Consume new data from the topic
+ config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();
+ return new OpaqueTridentKafkaSpout(config);
+ }
+
private Stream addDRPCStream(TridentTopology tridentTopology, TridentState state, LocalDRPC drpc) {
return tridentTopology.newDRPCStream("words", drpc)
@@ -111,12 +127,14 @@ public class TridentKafkaWordCount {
.project(new Fields("word", "count"));
}
- private TridentState addTridentState(TridentTopology tridentTopology) {
- return tridentTopology.newStream("spout1", createKafkaSpout()).parallelismHint(1)
+ protected TridentState addTridentState(TridentTopology tridentTopology) {
+// final Stream spoutStream = tridentTopology.newStream("spout1", createTransactionalKafkaSpout()).parallelismHint(1);
+ final Stream spoutStream = tridentTopology.newStream("spout1", createOpaqueKafkaSpout()).parallelismHint(1);
+
+ return spoutStream.each(spoutStream.getOutputFields(), new Debug(true))
.each(new Fields("str"), new Split(), new Fields("word"))
.groupBy(new Fields("word"))
- .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
- .parallelismHint(1);
+ .persistentAggregate(new DebugMemoryMapState.Factory(), new Count(), new Fields("count"));
}
/**
@@ -140,6 +158,7 @@ public class TridentKafkaWordCount {
public Config getConsumerConfig() {
Config conf = new Config();
conf.setMaxSpoutPending(20);
+ conf.setMaxTaskParallelism(1);
// conf.setDebug(true);
return conf;
}
@@ -152,7 +171,7 @@ public class TridentKafkaWordCount {
*/
public StormTopology buildProducerTopology(Properties prop) {
TopologyBuilder builder = new TopologyBuilder();
- builder.setSpout("spout", new RandomSentenceSpout(), 2);
+ builder.setSpout("spout", new RandomSentenceSpout.TimeStamped(""), 2);
/**
* The output field of the RandomSentenceSpout ("word") is provided as the boltMessageField
* so that this gets written out as the message in the kafka topic.
@@ -203,7 +222,6 @@ public class TridentKafkaWordCount {
* (word counts) by running an external drpc query against the drpc server.
*/
public static void main(String[] args) throws Exception {
-
String zkUrl = "localhost:2181"; // the defaults.
String brokerUrl = "localhost:9092";
@@ -213,20 +231,22 @@ public class TridentKafkaWordCount {
System.exit(1);
} else if (args.length == 1) {
zkUrl = args[0];
- } else {
+ } else if (args.length == 2) {
zkUrl = args[0];
brokerUrl = args[1];
}
System.out.println("Using Kafka zookeeper url: " + zkUrl + " broker url: " + brokerUrl);
- TridentKafkaWordCount wordCount = new TridentKafkaWordCount(zkUrl, brokerUrl);
+ runMain(args, new TridentKafkaWordCount(zkUrl, brokerUrl));
+ }
+ protected static void runMain(String[] args, TridentKafkaWordCount wordCount) throws Exception {
if (args.length == 3) {
Config conf = new Config();
conf.setMaxSpoutPending(20);
conf.setNumWorkers(1);
- // submit the consumer topology.
+ // submit the CONSUMER topology.
StormSubmitter.submitTopology(args[2] + "-consumer", conf, wordCount.buildConsumerTopology(null));
// submit the producer topology.
StormSubmitter.submitTopology(args[2] + "-producer", conf, wordCount.buildProducerTopology(wordCount.getProducerConfig()));
@@ -234,17 +254,19 @@ public class TridentKafkaWordCount {
LocalDRPC drpc = new LocalDRPC();
LocalCluster cluster = new LocalCluster();
- // submit the consumer topology.
+ // submit the CONSUMER topology.
cluster.submitTopology("wordCounter", wordCount.getConsumerConfig(), wordCount.buildConsumerTopology(drpc));
+ // submit the PRODUCER topology.
Config conf = new Config();
conf.setMaxSpoutPending(20);
- // submit the producer topology.
+// conf.setDebug(true);
cluster.submitTopology("kafkaBolt", conf, wordCount.buildProducerTopology(wordCount.getProducerConfig()));
// keep querying the word counts for a minute.
for (int i = 0; i < 60; i++) {
- System.out.println("DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
+ LOG.info("--- DRPC RESULT: " + drpc.execute("words", "the and apple snow jumped"));
+ System.out.println();
Thread.sleep(1000);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
index 822912b..8c22118 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaUtils.java
@@ -32,7 +32,13 @@ import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.UnresolvedAddressException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
@@ -215,6 +221,7 @@ public class KafkaUtils {
} else {
msgs = fetchResponse.messageSet(topic, partitionId);
}
+ LOG.debug("Messages fetched. [config = {}], [consumer = {}], [partition = {}], [offset = {}], [msgs = {}]", config, consumer, partition, offset, msgs);
return msgs;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java
index ac5b49f..7ce8d52 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TransactionalTridentKafkaSpout.java
@@ -17,13 +17,12 @@
*/
package org.apache.storm.kafka.trident;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.tuple.Fields;
import org.apache.storm.kafka.Partition;
+import org.apache.storm.task.TopologyContext;
import org.apache.storm.trident.spout.IPartitionedTridentSpout;
+import org.apache.storm.tuple.Fields;
import java.util.Map;
-import java.util.UUID;
public class TransactionalTridentKafkaSpout implements IPartitionedTridentSpout<GlobalPartitionInformation, Partition, Map> {
@@ -42,8 +41,7 @@ public class TransactionalTridentKafkaSpout implements IPartitionedTridentSpout<
@Override
public IPartitionedTridentSpout.Emitter getEmitter(Map conf, TopologyContext context) {
- return new TridentKafkaEmitter(conf, context, _config, context
- .getStormId()).asTransactionalEmitter();
+ return new TridentKafkaEmitter(conf, context, _config, context.getStormId()).asTransactionalEmitter();
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
index 809ed73..136eb0b 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
@@ -20,7 +20,13 @@ package org.apache.storm.kafka.trident;
import com.google.common.collect.ImmutableMap;
import org.apache.storm.Config;
-import org.apache.storm.kafka.*;
+import org.apache.storm.kafka.DynamicPartitionConnections;
+import org.apache.storm.kafka.FailedFetchException;
+import org.apache.storm.kafka.KafkaUtils;
+import org.apache.storm.kafka.MessageMetadataSchemeAsMultiScheme;
+import org.apache.storm.kafka.Partition;
+import org.apache.storm.kafka.PartitionManager;
+import org.apache.storm.kafka.TopicOffsetOutOfRangeException;
import org.apache.storm.metric.api.CombinedMetric;
import org.apache.storm.metric.api.MeanReducer;
import org.apache.storm.metric.api.ReducedMetric;
@@ -32,7 +38,11 @@ import org.apache.storm.trident.topology.TransactionAttempt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
@@ -65,7 +75,7 @@ public class TridentKafkaEmitter {
private Map failFastEmitNewPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map lastMeta) {
SimpleConsumer consumer = _connections.register(partition);
- Map ret = doEmitNewPartitionBatch(consumer, partition, collector, lastMeta);
+ Map ret = doEmitNewPartitionBatch(consumer, partition, collector, lastMeta, attempt);
Long offset = (Long) ret.get("offset");
Long endOffset = (Long) ret.get("nextOffset");
_kafkaOffsetMetric.setOffsetData(partition, new PartitionManager.OffsetData(endOffset, offset));
@@ -92,7 +102,8 @@ public class TridentKafkaEmitter {
}
}
- private Map doEmitNewPartitionBatch(SimpleConsumer consumer, Partition partition, TridentCollector collector, Map lastMeta) {
+ private Map doEmitNewPartitionBatch(SimpleConsumer consumer, Partition partition, TridentCollector collector, Map lastMeta, TransactionAttempt attempt) {
+ LOG.debug("Emitting new partition batch - [transaction = {}], [lastMeta = {}]", attempt, lastMeta);
long offset;
if (lastMeta != null) {
String lastInstanceId = null;
@@ -108,6 +119,7 @@ public class TridentKafkaEmitter {
} else {
offset = KafkaUtils.getOffset(consumer, partition.topic, partition.partition, _config);
}
+ LOG.debug("[transaction = {}], [OFFSET = {}]", attempt, offset);
ByteBufferMessageSet msgs = null;
try {
@@ -121,7 +133,7 @@ public class TridentKafkaEmitter {
long endoffset = offset;
for (MessageAndOffset msg : msgs) {
- emit(collector, msg.message(), partition, msg.offset());
+ emit(collector, msg.message(), partition, msg.offset(), attempt);
endoffset = msg.nextOffset();
}
Map newMeta = new HashMap();
@@ -132,6 +144,7 @@ public class TridentKafkaEmitter {
newMeta.put("broker", ImmutableMap.of("host", partition.host.host, "port", partition.host.port));
newMeta.put("topic", partition.topic);
newMeta.put("topology", ImmutableMap.of("name", _topologyName, "id", _topologyInstanceId));
+ LOG.debug("[transaction = {}], [newMeta = {}]", attempt, newMeta);
return newMeta;
}
@@ -172,14 +185,14 @@ public class TridentKafkaEmitter {
if (offset > nextOffset) {
throw new RuntimeException("Error when re-emitting batch. overshot the end offset");
}
- emit(collector, msg.message(), partition, msg.offset());
+ emit(collector, msg.message(), partition, msg.offset(), attempt);
offset = msg.nextOffset();
}
}
}
}
- private void emit(TridentCollector collector, Message msg, Partition partition, long offset) {
+ private void emit(TridentCollector collector, Message msg, Partition partition, long offset, TransactionAttempt attempt) {
Iterable<List<Object>> values;
if (_config.scheme instanceof MessageMetadataSchemeAsMultiScheme) {
values = KafkaUtils.generateTuples((MessageMetadataSchemeAsMultiScheme) _config.scheme, msg, partition, offset);
@@ -189,8 +202,11 @@ public class TridentKafkaEmitter {
if (values != null) {
for (List<Object> value : values) {
+ LOG.debug("Emitting: [Transaction: {}], [Data: {}]", attempt, value);
collector.emit(value);
}
+ } else {
+ LOG.debug("NOT Emitting NULL data. [Transaction: {}]", attempt);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/operation/builtin/Debug.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/operation/builtin/Debug.java b/storm-core/src/jvm/org/apache/storm/trident/operation/builtin/Debug.java
index 87c4167..a9f30d0 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/operation/builtin/Debug.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/operation/builtin/Debug.java
@@ -19,6 +19,8 @@ package org.apache.storm.trident.operation.builtin;
import org.apache.storm.trident.operation.BaseFilter;
import org.apache.storm.trident.tuple.TridentTuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Date;
@@ -26,10 +28,18 @@ import java.util.Date;
* Filter for debugging purposes. The `isKeep()` method simply prints the tuple to `System.out` and returns `true`.
*/
public class Debug extends BaseFilter {
+ private final Logger LOG = LoggerFactory.getLogger(Debug.class);
+
private final String name;
+ private boolean useLogger;
public Debug() {
- name = "DEBUG: ";
+ this(false);
+ }
+
+ public Debug(boolean useLogger) {
+ this.useLogger = useLogger;
+ this.name = "DEBUG: ";
}
/**
@@ -42,7 +52,11 @@ public class Debug extends BaseFilter {
@Override
public boolean isKeep(TridentTuple tuple) {
- System.out.println("<"+new Date()+"> "+name + tuple.toString());
+ if(useLogger) {
+ LOG.debug(tuple.toString());
+ } else {
+ System.out.println("<"+new Date()+"> "+name + tuple.toString());
+ }
return true;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java b/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
index bf4c093..fc4d5ea 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
@@ -18,12 +18,12 @@
package org.apache.storm.trident.spout;
import org.apache.storm.task.TopologyContext;
+import org.apache.storm.trident.operation.TridentCollector;
+import org.apache.storm.trident.topology.TransactionAttempt;
import org.apache.storm.tuple.Fields;
-import java.io.Serializable;
+
import java.util.List;
import java.util.Map;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.topology.TransactionAttempt;
/**
* This defines a transactional spout which does *not* necessarily
@@ -32,13 +32,14 @@ import org.apache.storm.trident.topology.TransactionAttempt;
*/
public interface IOpaquePartitionedTridentSpout<Partitions, Partition extends ISpoutPartition, M>
extends ITridentDataSource {
- public interface Coordinator<Partitions> {
+
+ interface Coordinator<Partitions> {
boolean isReady(long txid);
Partitions getPartitionsForBatch();
void close();
}
- public interface Emitter<Partitions, Partition extends ISpoutPartition, M> {
+ interface Emitter<Partitions, Partition extends ISpoutPartition, M> {
/**
* Emit a batch of tuples for a partition/transaction.
*
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/spout/IPartitionedTridentSpout.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/IPartitionedTridentSpout.java b/storm-core/src/jvm/org/apache/storm/trident/spout/IPartitionedTridentSpout.java
index 7450a3c..82c4f74 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/IPartitionedTridentSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/IPartitionedTridentSpout.java
@@ -18,12 +18,12 @@
package org.apache.storm.trident.spout;
import org.apache.storm.task.TopologyContext;
+import org.apache.storm.trident.operation.TridentCollector;
+import org.apache.storm.trident.topology.TransactionAttempt;
import org.apache.storm.tuple.Fields;
-import java.io.Serializable;
+
import java.util.List;
import java.util.Map;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.topology.TransactionAttempt;
/**
* This interface defines a transactional spout that reads its tuples from a partitioned set of
@@ -31,7 +31,7 @@ import org.apache.storm.trident.topology.TransactionAttempt;
* is always emitted for the same transaction id. The partition metadata is stored in Zookeeper.
*/
public interface IPartitionedTridentSpout<Partitions, Partition extends ISpoutPartition, T> extends ITridentDataSource {
- public interface Coordinator<Partitions> {
+ interface Coordinator<Partitions> {
/**
* Return the partitions currently in the source of data. The idea is
* is that if a new partition is added and a prior transaction is replayed, it doesn't
@@ -45,7 +45,7 @@ public interface IPartitionedTridentSpout<Partitions, Partition extends ISpoutPa
void close();
}
- public interface Emitter<Partitions, Partition extends ISpoutPartition, X> {
+ interface Emitter<Partitions, Partition extends ISpoutPartition, X> {
List<Partition> getOrderedPartitions(Partitions allPartitionInfo);
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java b/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
index 7c43961..ea66acd 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
@@ -19,7 +19,14 @@ package org.apache.storm.trident.spout;
import org.apache.storm.task.TopologyContext;
+import org.apache.storm.trident.operation.TridentCollector;
+import org.apache.storm.trident.topology.TransactionAttempt;
+import org.apache.storm.trident.topology.state.RotatingTransactionalState;
+import org.apache.storm.trident.topology.state.TransactionalState;
import org.apache.storm.tuple.Fields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -28,13 +35,11 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.topology.state.RotatingTransactionalState;
-import org.apache.storm.trident.topology.state.TransactionalState;
-import org.apache.storm.trident.topology.TransactionAttempt;
public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentSpout<Object> {
+ protected final Logger LOG = LoggerFactory.getLogger(OpaquePartitionedTridentSpoutExecutor.class);
+
IOpaquePartitionedTridentSpout<Object, ISpoutPartition, Object> _spout;
public class Coordinator implements ITridentSpout.BatchCoordinator<Object> {
@@ -46,21 +51,28 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
@Override
public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
+ LOG.debug("Initialize Transaction. [txid = {}], [prevMetadata = {}], [currMetadata = {}]", txid, prevMetadata, currMetadata);
return _coordinator.getPartitionsForBatch();
}
+
@Override
public void close() {
+ LOG.debug("Closing");
_coordinator.close();
+ LOG.debug("Closed");
}
@Override
public void success(long txid) {
+ LOG.debug("Success [txid = {}]", txid);
}
@Override
public boolean isReady(long txid) {
- return _coordinator.isReady(txid);
+ boolean ready = _coordinator.isReady(txid);
+ LOG.debug("[isReady = {}], [txid = {}]", ready, txid);
+ return ready;
}
}
@@ -81,19 +93,23 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
Map<String, EmitterPartitionState> _partitionStates = new HashMap<>();
int _index;
int _numTasks;
-
+
public Emitter(String txStateId, Map conf, TopologyContext context) {
_emitter = _spout.getEmitter(conf, context);
_index = context.getThisTaskIndex();
_numTasks = context.getComponentTasks(context.getThisComponentId()).size();
- _state = TransactionalState.newUserState(conf, txStateId);
+ _state = TransactionalState.newUserState(conf, txStateId);
+ LOG.debug("Created {}", this);
}
-
+
Object _savedCoordinatorMeta = null;
boolean _changedMeta = false;
-
+
@Override
public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector) {
+ LOG.debug("Emitting Batch. [transaction = {}], [coordinatorMeta = {}], [collector = {}], [{}]",
+ tx, coordinatorMeta, collector, this);
+
if(_savedCoordinatorMeta==null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
_partitionStates.clear();
@@ -128,17 +144,21 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
Object meta = _emitter.emitPartitionBatch(tx, collector, s.partition, lastMeta);
metas.put(id, meta);
}
+ LOG.debug("Emitted Batch. [transaction = {}], [coordinatorMeta = {}], [collector = {}], [{}]",
+ tx, coordinatorMeta, collector, this);
}
@Override
public void success(TransactionAttempt tx) {
for(EmitterPartitionState state: _partitionStates.values()) {
state.rotatingState.cleanupBefore(tx.getTransactionId());
- }
+ }
+ LOG.debug("Success transaction {}. [{}]", tx, this);
}
@Override
public void commit(TransactionAttempt attempt) {
+ LOG.debug("Committing transaction {}. [{}]", attempt, this);
// this code here handles a case where a previous commit failed, and the partitions
// changed since the last commit. This clears out any state for the removed partitions
// for this txid.
@@ -159,21 +179,37 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
}
}
_changedMeta = false;
- }
+ }
Long txid = attempt.getTransactionId();
Map<String, Object> metas = _cachedMetas.remove(txid);
for(Entry<String, Object> entry: metas.entrySet()) {
_partitionStates.get(entry.getKey()).rotatingState.overrideState(txid, entry.getValue());
}
+ LOG.debug("Exiting commit method for transaction {}. [{}]", attempt, this);
}
@Override
public void close() {
+ LOG.debug("Closing");
_emitter.close();
+ LOG.debug("Closed");
}
- }
-
+
+ @Override
+ public String toString() {
+ return "Emitter{" +
+ ", _state=" + _state +
+ ", _cachedMetas=" + _cachedMetas +
+ ", _partitionStates=" + _partitionStates +
+ ", _index=" + _index +
+ ", _numTasks=" + _numTasks +
+ ", _savedCoordinatorMeta=" + _savedCoordinatorMeta +
+ ", _changedMeta=" + _changedMeta +
+ '}';
+ }
+ }
+
public OpaquePartitionedTridentSpoutExecutor(IOpaquePartitionedTridentSpout<Object, ISpoutPartition, Object> spout) {
_spout = spout;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java b/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
index dc9b3d9..a96dd1a 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
@@ -23,6 +23,8 @@ import org.apache.storm.trident.topology.TransactionAttempt;
import org.apache.storm.trident.topology.state.RotatingTransactionalState;
import org.apache.storm.trident.topology.state.TransactionalState;
import org.apache.storm.tuple.Fields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.HashMap;
@@ -31,6 +33,8 @@ import java.util.Map;
public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
+ private static final Logger LOG = LoggerFactory.getLogger(PartitionedTridentSpoutExecutor.class);
+
IPartitionedTridentSpout<Object, ISpoutPartition, Object> _spout;
public PartitionedTridentSpoutExecutor(IPartitionedTridentSpout<Object, ISpoutPartition, Object> spout) {
@@ -50,6 +54,8 @@ public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
@Override
public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
+ LOG.debug("Initialize Transaction. txid = {}, prevMetadata = {}, currMetadata = {}", txid, prevMetadata, currMetadata);
+
if(currMetadata!=null) {
return currMetadata;
} else {
@@ -60,16 +66,21 @@ public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
@Override
public void close() {
+ LOG.debug("Closing");
_coordinator.close();
+ LOG.debug("Closed");
}
@Override
public void success(long txid) {
+ LOG.debug("Success transaction id " + txid);
}
@Override
public boolean isReady(long txid) {
- return _coordinator.isReady(txid);
+ boolean ready = _coordinator.isReady(txid);
+ LOG.debug("isReady = {} ", ready);
+ return ready;
}
}
@@ -101,8 +112,9 @@ public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
@Override
- public void emitBatch(final TransactionAttempt tx, final Object coordinatorMeta,
- final TridentCollector collector) {
+ public void emitBatch(final TransactionAttempt tx, final Object coordinatorMeta, final TridentCollector collector) {
+ LOG.debug("Emitting Batch. [transaction = {}], [coordinatorMeta = {}], [collector = {}]", tx, coordinatorMeta, collector);
+
if(_savedCoordinatorMeta == null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
_partitionStates.clear();
@@ -133,11 +145,13 @@ public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
if(meta!=null) {
_emitter.emitPartitionBatch(tx, collector, partition, meta);
}
- }
+ }
+ LOG.debug("Emitted Batch. [tx = {}], [coordinatorMeta = {}], [collector = {}]", tx, coordinatorMeta, collector);
}
@Override
public void success(TransactionAttempt tx) {
+ LOG.debug("Success transaction " + tx);
for(EmitterPartitionState state: _partitionStates.values()) {
state.rotatingState.cleanupBefore(tx.getTransactionId());
}
@@ -145,8 +159,10 @@ public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
@Override
public void close() {
+ LOG.debug("Closing");
_state.close();
_emitter.close();
+ LOG.debug("Closed");
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/state/CombinerValueUpdater.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/state/CombinerValueUpdater.java b/storm-core/src/jvm/org/apache/storm/trident/state/CombinerValueUpdater.java
index 82b7360..f43071f 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/state/CombinerValueUpdater.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/state/CombinerValueUpdater.java
@@ -33,4 +33,13 @@ public class CombinerValueUpdater implements ValueUpdater<Object> {
if(stored==null) return arg;
else return agg.combine(stored, arg);
}
+
+ // helpful for debugging tests
+ public Object getArg() {
+ return arg;
+ }
+
+ public CombinerAggregator getAgg() {
+ return agg;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java b/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
index 6f59409..c10af43 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/topology/MasterBatchCoordinator.java
@@ -22,20 +22,20 @@ import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.trident.spout.ITridentSpout;
+import org.apache.storm.trident.topology.state.TransactionalState;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.WindowedTimeThrottler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;
-import java.util.Random;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.storm.trident.spout.ITridentSpout;
-import org.apache.storm.trident.topology.state.TransactionalState;
public class MasterBatchCoordinator extends BaseRichSpout {
public static final Logger LOG = LoggerFactory.getLogger(MasterBatchCoordinator.class);
@@ -74,6 +74,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
}
_managedSpoutIds = spoutIds;
_spouts = spouts;
+ LOG.debug("Created {}", this);
}
public List<String> getManagedSpoutIds(){
@@ -112,6 +113,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
String txId = _managedSpoutIds.get(i);
_coordinators.add(_spouts.get(i).getCoordinator(txId, conf, context));
}
+ LOG.debug("Opened {}", this);
}
@Override
@@ -119,6 +121,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
for(TransactionalState state: _states) {
state.close();
}
+ LOG.debug("Closed {}", this);
}
@Override
@@ -130,9 +133,11 @@ public class MasterBatchCoordinator extends BaseRichSpout {
public void ack(Object msgId) {
TransactionAttempt tx = (TransactionAttempt) msgId;
TransactionStatus status = _activeTx.get(tx.getTransactionId());
+ LOG.debug("Ack. [tx_attempt = {}], [tx_status = {}], [{}]", tx, status, this);
if(status!=null && tx.equals(status.attempt)) {
if(status.status==AttemptStatus.PROCESSING) {
status.status = AttemptStatus.PROCESSED;
+ LOG.debug("Changed status. [tx_attempt = {}] [tx_status = {}]", tx, status);
} else if(status.status==AttemptStatus.COMMITTING) {
_activeTx.remove(tx.getTransactionId());
_attemptIds.remove(tx.getTransactionId());
@@ -141,6 +146,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
for(TransactionalState state: _states) {
state.setData(CURRENT_TX, _currTransaction);
}
+ LOG.debug("Emitted on [stream = {}], [tx_attempt = {}], [tx_status = {}], [{}]", SUCCESS_STREAM_ID, tx, status, this);
}
sync();
}
@@ -150,6 +156,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
public void fail(Object msgId) {
TransactionAttempt tx = (TransactionAttempt) msgId;
TransactionStatus stored = _activeTx.remove(tx.getTransactionId());
+ LOG.debug("Fail. [tx_attempt = {}], [tx_status = {}], [{}]", tx, stored, this);
if(stored!=null && tx.equals(stored.attempt)) {
_activeTx.tailMap(tx.getTransactionId()).clear();
sync();
@@ -174,6 +181,7 @@ public class MasterBatchCoordinator extends BaseRichSpout {
if(maybeCommit!=null && maybeCommit.status == AttemptStatus.PROCESSED) {
maybeCommit.status = AttemptStatus.COMMITTING;
_collector.emit(COMMIT_STREAM_ID, new Values(maybeCommit.attempt), maybeCommit.attempt);
+ LOG.debug("Emitted on [stream = {}], [tx_status = {}], [{}]", COMMIT_STREAM_ID, maybeCommit, this);
}
if(_active) {
@@ -196,8 +204,10 @@ public class MasterBatchCoordinator extends BaseRichSpout {
}
TransactionAttempt attempt = new TransactionAttempt(curr, attemptId);
- _activeTx.put(curr, new TransactionStatus(attempt));
+ final TransactionStatus newTransactionStatus = new TransactionStatus(attempt);
+ _activeTx.put(curr, newTransactionStatus);
_collector.emit(BATCH_STREAM_ID, new Values(attempt), attempt);
+ LOG.debug("Emitted on [stream = {}], [tx_attempt = {}], [tx_status = {}], [{}]", BATCH_STREAM_ID, attempt, newTransactionStatus, this);
_throttler.markEvent();
}
curr = nextTransactionId(curr);
@@ -286,4 +296,21 @@ public class MasterBatchCoordinator extends BaseRichSpout {
ret.tailMap(currTransaction + maxBatches - 1).clear();
return ret;
}
+
+ @Override
+ public String toString() {
+ return "MasterBatchCoordinator{" +
+ "_states=" + _states +
+ ", _activeTx=" + _activeTx +
+ ", _attemptIds=" + _attemptIds +
+ ", _collector=" + _collector +
+ ", _currTransaction=" + _currTransaction +
+ ", _maxTransactionActive=" + _maxTransactionActive +
+ ", _coordinators=" + _coordinators +
+ ", _managedSpoutIds=" + _managedSpoutIds +
+ ", _spouts=" + _spouts +
+ ", _throttler=" + _throttler +
+ ", _active=" + _active +
+ "}";
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/topology/state/RotatingTransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/topology/state/RotatingTransactionalState.java b/storm-core/src/jvm/org/apache/storm/trident/topology/state/RotatingTransactionalState.java
index ebd73b2..6078f51 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/topology/state/RotatingTransactionalState.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/topology/state/RotatingTransactionalState.java
@@ -19,6 +19,8 @@ package org.apache.storm.trident.topology.state;
import org.apache.storm.utils.Utils;
import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.List;
@@ -26,6 +28,8 @@ import java.util.SortedMap;
import java.util.TreeMap;
public class RotatingTransactionalState {
+ private static final Logger LOG = LoggerFactory.getLogger(RotatingTransactionalState.class);
+
public static interface StateInitializer {
Object init(long txid, Object lastState);
}
@@ -40,6 +44,7 @@ public class RotatingTransactionalState {
_subdir = subdir;
state.mkdir(subdir);
sync();
+ LOG.debug("Created {}", this);
}
@@ -49,19 +54,30 @@ public class RotatingTransactionalState {
}
public void overrideState(long txid, Object state) {
+ LOG.debug("Overriding state. [txid = {}], [state = {}]", txid, state);
+ LOG.trace("[{}]", this);
+
_state.setData(txPath(txid), state);
_curr.put(txid, state);
+
+ LOG.trace("Overriding state complete. [{}]", this);
}
public void removeState(long txid) {
+ Object state = null;
if(_curr.containsKey(txid)) {
- _curr.remove(txid);
+ state = _curr.remove(txid);
_state.delete(txPath(txid));
}
+ LOG.debug("Removed [state = {}], [txid = {}]", state, txid);
+ LOG.trace("[{}]", this);
}
public Object getState(long txid) {
- return _curr.get(txid);
+ final Object state = _curr.get(txid);
+ LOG.debug("Getting state. [txid = {}] => [state = {}]", txid, state);
+ LOG.trace("Internal state [{}]", this);
+ return state;
}
public Object getState(long txid, StateInitializer init) {
@@ -87,13 +103,26 @@ public class RotatingTransactionalState {
_curr.put(txid, data);
_state.setData(txPath(txid), data);
}
- return _curr.get(txid);
+ Object state = _curr.get(txid);
+ LOG.debug("Getting or initializing state. [txid = {}] => [state = {}]", txid, state);
+ LOG.trace("[{}]", this);
+ return state;
}
public Object getPreviousState(long txid) {
- SortedMap<Long, Object> prevMap = _curr.headMap(txid);
- if(prevMap.isEmpty()) return null;
- else return prevMap.get(prevMap.lastKey());
+ final SortedMap<Long, Object> prevMap = _curr.headMap(txid);
+ Object state;
+
+ if(prevMap.isEmpty()) {
+ state = null;
+ }
+ else {
+ state = prevMap.get(prevMap.lastKey());
+ }
+
+ LOG.debug("Getting previous [state = {}], [txid = {}]", state, txid);
+ LOG.trace("[{}]", this);
+ return state;
}
public boolean hasCache(long txid) {
@@ -104,12 +133,14 @@ public class RotatingTransactionalState {
* Returns null if it was created, the value otherwise.
*/
public Object getStateOrCreate(long txid, StateInitializer init) {
+ Object state;
if(_curr.containsKey(txid)) {
- return _curr.get(txid);
+ state = _curr.get(txid);
} else {
getState(txid, init);
- return null;
+ state = null;
}
+ return state;
}
public void cleanupBefore(long txid) {
@@ -142,6 +173,14 @@ public class RotatingTransactionalState {
private String txPath(String tx) {
return _subdir + "/" + tx;
- }
-
+ }
+
+ @Override
+ public String toString() {
+ return "RotatingTransactionalState{" +
+ "_state=" + _state +
+ ", _subdir='" + _subdir + '\'' +
+ ", _curr=" + _curr +
+ '}';
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/91903743/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java b/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
index 3bf2794..af1f297 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
@@ -18,27 +18,28 @@
package org.apache.storm.trident.topology.state;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.PathAndBytesable;
+import org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
import org.apache.storm.Config;
import org.apache.storm.utils.Utils;
import org.apache.storm.utils.ZookeeperAuthInfo;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
-import org.apache.curator.framework.api.PathAndBytesable;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
-import org.json.simple.JSONValue;
public class TransactionalState {
+ private static final Logger LOG = LoggerFactory.getLogger(TransactionalState.class);
+
CuratorFramework _curator;
List<ACL> _zkAcls = null;
@@ -87,6 +88,7 @@ public class TransactionalState {
byte[] data, List<ACL> acls, CreateMode mode) throws Exception {
ProtectACLCreateModePathAndBytesable<String> builder =
curator.create().creatingParentsIfNeeded();
+ LOG.debug("Creating node [path = {}], [data = {}], [acls = {}], [mode = {}]", path, asString(data), acls, mode);
if (acls == null) {
if (mode == null ) {
@@ -100,6 +102,10 @@ public class TransactionalState {
TransactionalState.forPath(builder.withACL(acls), path, data);
}
+ private static String asString(byte[] data) {
+ return data == null ? "null" : new String(data);
+ }
+
public void setData(String path, Object obj) {
path = "/" + path;
byte[] ser;
@@ -115,6 +121,7 @@ public class TransactionalState {
TransactionalState.createNode(_curator, path, ser, _zkAcls,
CreateMode.PERSISTENT);
}
+ LOG.debug("Set [path = {}] => [data = {}]", path, asString(ser));
} catch(Exception e) {
throw new RuntimeException(e);
}
@@ -127,19 +134,23 @@ public class TransactionalState {
} catch (Exception e) {
throw new RuntimeException(e);
}
+ LOG.debug("Deleted [path = {}]", path);
}
public List<String> list(String path) {
path = "/" + path;
try {
+ List<String> children;
if(_curator.checkExists().forPath(path)==null) {
- return new ArrayList<String>();
+ children = new ArrayList<>();
} else {
- return _curator.getChildren().forPath(path);
+ children = _curator.getChildren().forPath(path);
}
+ LOG.debug("List [path = {}], [children = {}]", path, children);
+ return children;
} catch(Exception e) {
throw new RuntimeException(e);
- }
+ }
}
public void mkdir(String path) {
@@ -149,11 +160,14 @@ public class TransactionalState {
public Object getData(String path) {
path = "/" + path;
try {
+ Object data;
if(_curator.checkExists().forPath(path)!=null) {
- return JSONValue.parse(new String(_curator.getData().forPath(path), "UTF-8"));
+ data = JSONValue.parse(new String(_curator.getData().forPath(path), "UTF-8"));
} else {
- return null;
+ data = null;
}
+ LOG.debug("Get. [path = {}] => [data = {}]", path, data);
+ return data;
} catch(Exception e) {
throw new RuntimeException(e);
}
[2/2] storm git commit: Added STORM-2097 to CHANGELOG.
Posted by sr...@apache.org.
Added STORM-2097 to CHANGELOG.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d7d7e61a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d7d7e61a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d7d7e61a
Branch: refs/heads/1.x-branch
Commit: d7d7e61a152b83d4acc1b37d280e22795088fd9f
Parents: 9190374
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Tue Oct 11 09:07:00 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Tue Oct 11 09:07:00 2016 -0700
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d7d7e61a/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 577158b..8b8efba 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.1.0
+ * STORM-2097: Improve logging in trident core and examples
* STORM-2144: Fix Storm-sql group-by behavior in standalone mode
* STORM-2066: make error message in IsolatedPool.java more descriptive
* STORM-1870: Allow FluxShellBolt/Spout set custom "componentConfig" via yaml