use sync send method Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/88f71b85 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/88f71b85 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/88f71b85
Branch: refs/heads/master Commit: 88f71b858e2589abbd477f560bfbb706225ea82c Parents: 5c38b91 Author: Xin Wang <[email protected]> Authored: Thu Oct 29 13:09:20 2015 +0800 Committer: Xin Wang <[email protected]> Committed: Thu Oct 29 13:09:20 2015 +0800 ---------------------------------------------------------------------- .../storm/kafka/trident/TridentKafkaState.java | 36 +++++++++----------- 1 file changed, 17 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/88f71b85/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java ---------------------------------------------------------------------- diff --git a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java index 098805c..7f70ce4 100644 --- a/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java +++ b/external/storm-kafka/src/jvm/storm/kafka/trident/TridentKafkaState.java @@ -20,10 +20,8 @@ package storm.kafka.trident; import backtype.storm.task.OutputCollector; import backtype.storm.topology.FailedException; import org.apache.commons.lang.Validate; -import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import storm.kafka.trident.mapper.TridentTupleToKafkaMapper; @@ -77,24 +75,24 @@ public class TridentKafkaState implements State { } public void updateState(List<TridentTuple> tuples, TridentCollector collector) { - for (final TridentTuple tuple : tuples) { - final String topic = topicSelector.getTopic(tuple); - if(topic != null) { - producer.send(new ProducerRecord(topic, mapper.getKeyFromTuple(tuple), - mapper.getMessageFromTuple(tuple)),new Callback() { - @Override - public void onCompletion(RecordMetadata metadata, Exception ex) { - if(ex != null){ - String errorMsg = "Could not send message with key = " - + mapper.getKeyFromTuple(tuple) + " to topic = " + topic; - LOG.warn(errorMsg, ex); - throw new FailedException(errorMsg, ex); - } - } - }); - } else { - LOG.warn("skipping key = " + mapper.getKeyFromTuple(tuple) + ", topic selector returned null."); + String topic = null; + for (TridentTuple tuple : tuples) { + try { + topic = topicSelector.getTopic(tuple); + + if(topic != null) { + producer.send(new ProducerRecord(topic, mapper.getKeyFromTuple(tuple), + mapper.getMessageFromTuple(tuple))).get(); + } else { + LOG.warn("skipping key = " + mapper.getKeyFromTuple(tuple) + ", topic selector returned null."); + } + } catch (Exception ex) { + String errorMsg = "Could not send message with key = " + mapper.getKeyFromTuple(tuple) + + " to topic = " + topic; + LOG.warn(errorMsg, ex); + throw new FailedException(errorMsg, ex); } } } + } }
