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 2017/12/22 17:45:12 UTC
storm git commit: STORM-2864: Minor optimisation about trident kafka
state Make TridentKafkaState a template class to eliminate warning messages
in eclipse,
and a minor optimization that use StringBuilder.append instead of string
concat operation.
Repository: storm
Updated Branches:
refs/heads/1.x-branch e296d65fd -> 7426c9874
STORM-2864: Minor optimisation about trident kafka state
Make TridentKafkaState a template class to eliminate warning messages in
eclipse, and a minor optimization that use StringBuilder.append instead
of string concat operation.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/7426c987
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/7426c987
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/7426c987
Branch: refs/heads/1.x-branch
Commit: 7426c9874e9afb971a32da77f4e05c65c9078bd3
Parents: e296d65
Author: OuYang Liang <ou...@gmail.com>
Authored: Wed Dec 20 22:51:29 2017 +0800
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Fri Dec 22 18:43:11 2017 +0100
----------------------------------------------------------------------
.../storm/kafka/trident/TridentKafkaState.java | 70 ++++++++++----------
.../kafka/trident/TridentKafkaStateFactory.java | 17 ++---
.../kafka/trident/TridentKafkaStateUpdater.java | 6 +-
3 files changed, 48 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/7426c987/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaState.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaState.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaState.java
index 44b6d5f..2e2c13b 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaState.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaState.java
@@ -37,21 +37,20 @@ import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-public class TridentKafkaState implements State {
+public class TridentKafkaState<K, V> implements State {
private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaState.class);
- private KafkaProducer producer;
- private OutputCollector collector;
+ private KafkaProducer<K, V> producer;
- private TridentTupleToKafkaMapper mapper;
+ private TridentTupleToKafkaMapper<K, V> mapper;
private KafkaTopicSelector topicSelector;
- public TridentKafkaState withTridentTupleToKafkaMapper(TridentTupleToKafkaMapper mapper) {
+ public TridentKafkaState<K, V> withTridentTupleToKafkaMapper(TridentTupleToKafkaMapper<K, V> mapper) {
this.mapper = mapper;
return this;
}
- public TridentKafkaState withKafkaTopicSelector(KafkaTopicSelector selector) {
+ public TridentKafkaState<K, V> withKafkaTopicSelector(KafkaTopicSelector selector) {
this.topicSelector = selector;
return this;
}
@@ -73,7 +72,7 @@ public class TridentKafkaState implements State {
public void prepare(Properties options) {
Objects.requireNonNull(mapper, "mapper can not be null");
Objects.requireNonNull(topicSelector, "topicSelector can not be null");
- producer = new KafkaProducer(options);
+ producer = new KafkaProducer<>(options);
}
/**
@@ -85,26 +84,26 @@ public class TridentKafkaState implements State {
String topic = null;
try {
long startTime = System.currentTimeMillis();
- int numberOfRecords = tuples.size();
- List<Future<RecordMetadata>> futures = new ArrayList<>(numberOfRecords);
+ int numberOfRecords = tuples.size();
+ List<Future<RecordMetadata>> futures = new ArrayList<>(numberOfRecords);
for (TridentTuple tuple : tuples) {
topic = topicSelector.getTopic(tuple);
- Object messageFromTuple = mapper.getMessageFromTuple(tuple);
- Object keyFromTuple = mapper.getKeyFromTuple(tuple);
-
+ V messageFromTuple = mapper.getMessageFromTuple(tuple);
+ K keyFromTuple = mapper.getKeyFromTuple(tuple);
+
if (topic != null) {
- if (messageFromTuple != null) {
- Future<RecordMetadata> result = producer.send(new ProducerRecord(topic,keyFromTuple, messageFromTuple));
- futures.add(result);
- } else {
- LOG.warn("skipping Message with Key "+ keyFromTuple +" as message was null");
- }
-
+ if (messageFromTuple != null) {
+ Future<RecordMetadata> result = producer.send(new ProducerRecord<>(topic, keyFromTuple, messageFromTuple));
+ futures.add(result);
+ } else {
+ LOG.warn("skipping Message with Key {} as message was null", keyFromTuple);
+ }
+
} else {
- LOG.warn("skipping key = " + keyFromTuple + ", topic selector returned null.");
+ LOG.warn("skipping key = {}, topic selector returned null.", keyFromTuple);
}
}
-
+
int emittedRecords = futures.size();
List<ExecutionException> exceptions = new ArrayList<>(emittedRecords);
for (Future<RecordMetadata> future : futures) {
@@ -115,20 +114,21 @@ public class TridentKafkaState implements State {
}
}
- if (exceptions.size() > 0){
- StringBuilder errorMsg = new StringBuilder("Could not retrieve result for messages " + tuples + " from topic = " + topic
- + " because of the following exceptions:" + System.lineSeparator());
-
- for (ExecutionException exception : exceptions) {
- errorMsg = errorMsg.append(exception.getMessage()).append(System.lineSeparator()); ;
- }
- String message = errorMsg.toString();
- LOG.error(message);
- throw new FailedException(message);
- }
- long latestTime = System.currentTimeMillis();
- LOG.info("Emitted record {} sucessfully in {} ms to topic {} ", emittedRecords, latestTime-startTime, topic);
-
+ if (exceptions.size() > 0) {
+ StringBuilder errorMsg = new StringBuilder("Could not retrieve result for messages ");
+ errorMsg.append(tuples).append(" from topic = ").append(topic)
+ .append(" because of the following exceptions:").append(System.lineSeparator());
+
+ for (ExecutionException exception : exceptions) {
+ errorMsg = errorMsg.append(exception.getMessage()).append(System.lineSeparator());
+ }
+ String message = errorMsg.toString();
+ LOG.error(message);
+ throw new FailedException(message);
+ }
+ long latestTime = System.currentTimeMillis();
+ LOG.info("Emitted record {} sucessfully in {} ms to topic {} ", emittedRecords, latestTime - startTime, topic);
+
} catch (Exception ex) {
String errorMsg = "Could not send messages " + tuples + " to topic = " + topic;
LOG.warn(errorMsg, ex);
http://git-wip-us.apache.org/repos/asf/storm/blob/7426c987/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateFactory.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateFactory.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateFactory.java
index f564510..35620de 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateFactory.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateFactory.java
@@ -28,25 +28,26 @@ import org.apache.storm.trident.state.StateFactory;
import java.util.Map;
import java.util.Properties;
-public class TridentKafkaStateFactory implements StateFactory {
+public class TridentKafkaStateFactory<K, V> implements StateFactory {
+ private static final long serialVersionUID = -3613240970062343385L;
private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaStateFactory.class);
- private TridentTupleToKafkaMapper mapper;
+ private TridentTupleToKafkaMapper<K, V> mapper;
private KafkaTopicSelector topicSelector;
private Properties producerProperties = new Properties();
- public TridentKafkaStateFactory withTridentTupleToKafkaMapper(TridentTupleToKafkaMapper mapper) {
+ public TridentKafkaStateFactory<K, V> withTridentTupleToKafkaMapper(TridentTupleToKafkaMapper<K, V> mapper) {
this.mapper = mapper;
return this;
}
- public TridentKafkaStateFactory withKafkaTopicSelector(KafkaTopicSelector selector) {
+ public TridentKafkaStateFactory<K, V> withKafkaTopicSelector(KafkaTopicSelector selector) {
this.topicSelector = selector;
return this;
}
- public TridentKafkaStateFactory withProducerProperties(Properties props) {
+ public TridentKafkaStateFactory<K, V> withProducerProperties(Properties props) {
this.producerProperties = props;
return this;
}
@@ -54,9 +55,9 @@ public class TridentKafkaStateFactory implements StateFactory {
@Override
public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
LOG.info("makeState(partitonIndex={}, numpartitions={}", partitionIndex, numPartitions);
- TridentKafkaState state = new TridentKafkaState()
- .withKafkaTopicSelector(this.topicSelector)
- .withTridentTupleToKafkaMapper(this.mapper);
+ TridentKafkaState<K, V> state = new TridentKafkaState<>();
+ state.withKafkaTopicSelector(this.topicSelector)
+ .withTridentTupleToKafkaMapper(this.mapper);
state.prepare(producerProperties);
return state;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/7426c987/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateUpdater.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateUpdater.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateUpdater.java
index 89535c6..19e3d33 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateUpdater.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/trident/TridentKafkaStateUpdater.java
@@ -23,10 +23,12 @@ import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.state.BaseStateUpdater;
import org.apache.storm.trident.tuple.TridentTuple;
-public class TridentKafkaStateUpdater extends BaseStateUpdater<TridentKafkaState> {
+public class TridentKafkaStateUpdater<K, V> extends BaseStateUpdater<TridentKafkaState<K, V>> {
+
+ private static final long serialVersionUID = 3352659585225274402L;
@Override
- public void updateState(TridentKafkaState state, List<TridentTuple> tuples, TridentCollector collector) {
+ public void updateState(TridentKafkaState<K, V> state, List<TridentTuple> tuples, TridentCollector collector) {
state.updateState(tuples, collector);
}
}