You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/09/06 13:58:52 UTC
storm git commit: Merge branch 'druidlog' of
https://github.com/omkreddy/storm into PULL-2308-1.x
Repository: storm
Updated Branches:
refs/heads/master a9f92dba0 -> 9bf202277
Merge branch 'druidlog' of https://github.com/omkreddy/storm into PULL-2308-1.x
MINOR: Update DruidBeamBolt logs
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9bf20227
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9bf20227
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9bf20227
Branch: refs/heads/master
Commit: 9bf202277bf02cee8fbe8c627aa1834d00f556ef
Parents: a9f92db
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Wed Sep 6 08:20:42 2017 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Wed Sep 6 08:21:36 2017 -0500
----------------------------------------------------------------------
.../org/apache/storm/druid/bolt/DruidBeamBolt.java | 13 ++++++-------
1 file changed, 6 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9bf20227/external/storm-druid/src/main/java/org/apache/storm/druid/bolt/DruidBeamBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-druid/src/main/java/org/apache/storm/druid/bolt/DruidBeamBolt.java b/external/storm-druid/src/main/java/org/apache/storm/druid/bolt/DruidBeamBolt.java
index c2feb18..92b8a21 100644
--- a/external/storm-druid/src/main/java/org/apache/storm/druid/bolt/DruidBeamBolt.java
+++ b/external/storm-druid/src/main/java/org/apache/storm/druid/bolt/DruidBeamBolt.java
@@ -25,12 +25,10 @@ import com.twitter.util.FutureEventListener;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseTickTupleAwareRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.TupleUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -78,27 +76,28 @@ public class DruidBeamBolt<E> extends BaseTickTupleAwareRichBolt {
@Override
protected void process(final Tuple tuple) {
- Future future = tranquilizer.send((druidEventMapper.getEvent(tuple)));
- LOG.debug("Sent tuple : [{}]", tuple);
+ final E mappedEvent = druidEventMapper.getEvent(tuple);
+ Future future = tranquilizer.send(mappedEvent);
+ LOG.debug("Sent tuple : [{}]", mappedEvent);
future.addEventListener(new FutureEventListener() {
@Override
public void onFailure(Throwable cause) {
if (cause instanceof MessageDroppedException) {
collector.ack(tuple);
- LOG.debug("Tuple Dropped due to MessageDroppedException : [{}]", tuple);
+ LOG.debug("Tuple Dropped due to MessageDroppedException {} : [{}]", cause.getMessage(), mappedEvent);
if (druidConfig.getDiscardStreamId() != null)
collector.emit(druidConfig.getDiscardStreamId(), new Values(tuple, System.currentTimeMillis()));
} else {
collector.fail(tuple);
- LOG.debug("Tuple Processing Failed : [{}]", tuple);
+ LOG.error("Tuple Processing Failed : [{}]", mappedEvent, cause);
}
}
@Override
public void onSuccess(Object value) {
collector.ack(tuple);
- LOG.debug("Tuple Processing Success : [{}]", tuple);
+ LOG.debug("Tuple Processing Success : [{}]", mappedEvent);
}
});