You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2018/02/07 21:05:26 UTC
[4/7] storm git commit: STORM-2936 Overwrite latest
storm-kafka-client 1.x-branch into 1.1.x-branch
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
index 7dfe7f6..c9f9541 100755
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/internal/OffsetManager.java
@@ -13,13 +13,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.kafka.spout.internal;
+import com.google.common.annotations.VisibleForTesting;
+
import java.util.Comparator;
import java.util.Iterator;
import java.util.NavigableSet;
+import java.util.NoSuchElementException;
import java.util.TreeSet;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
+
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.storm.kafka.spout.KafkaSpoutMessageId;
@@ -30,23 +34,27 @@ import org.slf4j.LoggerFactory;
* Manages acked and committed offsets for a TopicPartition. This class is not thread safe
*/
public class OffsetManager {
-
private static final Comparator<KafkaSpoutMessageId> OFFSET_COMPARATOR = new OffsetComparator();
private static final Logger LOG = LoggerFactory.getLogger(OffsetManager.class);
+
private final TopicPartition tp;
- /* First offset to be fetched. It is either set to the beginning, end, or to the first uncommitted offset.
- * Initial value depends on offset strategy. See KafkaSpoutConsumerRebalanceListener */
- private final long initialFetchOffset;
- // Committed offset, i.e. the offset where processing will resume if the spout restarts. Initially it is set to fetchOffset.
- private long committedOffset;
// Emitted Offsets List
private final NavigableSet<Long> emittedOffsets = new TreeSet<>();
// Acked messages sorted by ascending order of offset
private final NavigableSet<KafkaSpoutMessageId> ackedMsgs = new TreeSet<>(OFFSET_COMPARATOR);
+ // Committed offset, i.e. the offset where processing will resume upon spout restart. Initially it is set to fetchOffset.
+ private long committedOffset;
+ // True if this OffsetManager has made at least one commit to Kafka
+ private boolean committed;
+ private long latestEmittedOffset;
+ /**
+ * Creates a new OffsetManager.
+ * @param tp The TopicPartition
+ * @param initialFetchOffset The initial fetch offset for the given TopicPartition
+ */
public OffsetManager(TopicPartition tp, long initialFetchOffset) {
this.tp = tp;
- this.initialFetchOffset = initialFetchOffset;
this.committedOffset = initialFetchOffset;
LOG.debug("Instantiated {}", this.toString());
}
@@ -56,7 +64,29 @@ public class OffsetManager {
}
public void addToEmitMsgs(long offset) {
- this.emittedOffsets.add(offset); // O(Log N)
+ this.emittedOffsets.add(offset); // O(Log N)
+ this.latestEmittedOffset = Math.max(latestEmittedOffset, offset);
+ }
+
+ public int getNumUncommittedOffsets() {
+ return this.emittedOffsets.size();
+ }
+
+ /**
+ * Gets the offset of the nth emitted message after the committed offset.
+ * Example: If the committed offset is 0 and offsets 1, 2, 8, 10 have been emitted,
+ * getNthUncommittedOffsetAfterCommittedOffset(3) returns 8.
+ *
+ * @param index The index of the message to get the offset for
+ * @return The offset
+ * @throws NoSuchElementException if the index is out of range
+ */
+ public long getNthUncommittedOffsetAfterCommittedOffset(int index) {
+ Iterator<Long> offsetIter = emittedOffsets.iterator();
+ for (int i = 0; i < index - 1; i++) {
+ offsetIter.next();
+ }
+ return offsetIter.next();
}
/**
@@ -66,20 +96,20 @@ public class OffsetManager {
* <p/>
* The returned offset points to the earliest uncommitted offset, and matches the semantics of the KafkaConsumer.commitSync API.
*
+ * @param commitMetadata Metadata information to commit to Kafka. It is constant per KafkaSpout instance per topology
* @return the next OffsetAndMetadata to commit, or null if no offset is
- * ready to commit.
+ * ready to commit.
*/
- public OffsetAndMetadata findNextCommitOffset() {
+ public OffsetAndMetadata findNextCommitOffset(final String commitMetadata) {
boolean found = false;
long currOffset;
long nextCommitOffset = committedOffset;
- KafkaSpoutMessageId nextCommitMsg = null; // this is a convenience variable to make it faster to create OffsetAndMetadata
for (KafkaSpoutMessageId currAckedMsg : ackedMsgs) { // complexity is that of a linear scan on a TreeMap
currOffset = currAckedMsg.offset();
- if (currOffset == nextCommitOffset) { // found the next offset to commit
+ if (currOffset == nextCommitOffset) {
+ // found the next offset to commit
found = true;
- nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset + 1;
} else if (currOffset > nextCommitOffset) {
if (emittedOffsets.contains(nextCommitOffset)) {
@@ -101,7 +131,6 @@ public class OffsetManager {
if (nextEmittedOffset != null && currOffset == nextEmittedOffset) {
LOG.debug("Found committable offset: [{}] after missing offset: [{}], skipping to the committable offset",
currOffset, nextCommitOffset);
- nextCommitMsg = currAckedMsg;
nextCommitOffset = currOffset + 1;
} else {
LOG.debug("Topic-partition [{}] has non-sequential offset [{}]."
@@ -118,10 +147,10 @@ public class OffsetManager {
OffsetAndMetadata nextCommitOffsetAndMetadata = null;
if (found) {
- nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset,
- nextCommitMsg.getMetadata(Thread.currentThread()));
+ nextCommitOffsetAndMetadata = new OffsetAndMetadata(nextCommitOffset, commitMetadata);
+
LOG.debug("Topic-partition [{}] has offsets [{}-{}] ready to be committed."
- + " Processing will resume at offset [{}] if the spout restarts",
+ + " Processing will resume at offset [{}] upon spout restart",
tp, committedOffset, nextCommitOffsetAndMetadata.offset() - 1, nextCommitOffsetAndMetadata.offset());
} else {
LOG.debug("Topic-partition [{}] has no offsets ready to be committed", tp);
@@ -133,18 +162,19 @@ public class OffsetManager {
/**
* Marks an offset as committed. This method has side effects - it sets the
* internal state in such a way that future calls to
- * {@link #findNextCommitOffset()} will return offsets greater than or equal to the
+ * {@link #findNextCommitOffset(String)} will return offsets greater than or equal to the
* offset specified, if any.
*
- * @param committedOffset The committed offset. All lower offsets are expected to have been committed.
+ * @param committedOffsetAndMeta The committed offset. All lower offsets are expected to have been committed.
* @return Number of offsets committed in this commit
*/
- public long commit(OffsetAndMetadata committedOffset) {
+ public long commit(OffsetAndMetadata committedOffsetAndMeta) {
+ committed = true;
final long preCommitCommittedOffset = this.committedOffset;
long numCommittedOffsets = 0;
- this.committedOffset = committedOffset.offset();
+ this.committedOffset = committedOffsetAndMeta.offset();
for (Iterator<KafkaSpoutMessageId> iterator = ackedMsgs.iterator(); iterator.hasNext();) {
- if (iterator.next().offset() < committedOffset.offset()) {
+ if (iterator.next().offset() < committedOffsetAndMeta.offset()) {
iterator.remove();
numCommittedOffsets++;
} else {
@@ -153,7 +183,7 @@ public class OffsetManager {
}
for (Iterator<Long> iterator = emittedOffsets.iterator(); iterator.hasNext();) {
- if (iterator.next() < committedOffset.offset()) {
+ if (iterator.next() < committedOffsetAndMeta.offset()) {
iterator.remove();
} else {
break;
@@ -163,41 +193,46 @@ public class OffsetManager {
LOG.trace("{}", this);
LOG.debug("Committed [{}] offsets in the range [{}-{}] for topic-partition [{}]."
- + " Processing will resume at [{}] if the spout restarts.",
+ + " Processing will resume at [{}] upon spout restart",
numCommittedOffsets, preCommitCommittedOffset, this.committedOffset - 1, tp, this.committedOffset);
return numCommittedOffsets;
}
- public long getCommittedOffset() {
- return committedOffset;
- }
-
- public boolean isEmpty() {
- return ackedMsgs.isEmpty();
- }
-
- public boolean contains(ConsumerRecord record) {
- return contains(new KafkaSpoutMessageId(record));
+ /**
+ * Checks if this OffsetManager has committed to Kafka.
+ *
+ * @return true if this OffsetManager has made at least one commit to Kafka, false otherwise
+ */
+ public boolean hasCommitted() {
+ return committed;
}
public boolean contains(KafkaSpoutMessageId msgId) {
return ackedMsgs.contains(msgId);
}
-
- //VisibleForTesting
+
+ @VisibleForTesting
boolean containsEmitted(long offset) {
return emittedOffsets.contains(offset);
}
+ public long getLatestEmittedOffset() {
+ return latestEmittedOffset;
+ }
+
+ public long getCommittedOffset() {
+ return committedOffset;
+ }
+
@Override
public final String toString() {
return "OffsetManager{"
+ "topic-partition=" + tp
- + ", fetchOffset=" + initialFetchOffset
+ ", committedOffset=" + committedOffset
+ ", emittedOffsets=" + emittedOffsets
+ ", ackedMsgs=" + ackedMsgs
+ + ", latestEmittedOffset=" + latestEmittedOffset
+ '}';
}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics/KafkaOffsetMetric.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics/KafkaOffsetMetric.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics/KafkaOffsetMetric.java
new file mode 100644
index 0000000..afe8f74
--- /dev/null
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics/KafkaOffsetMetric.java
@@ -0,0 +1,141 @@
+/**
+ * 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.kafka.spout.metrics;
+
+import com.google.common.base.Supplier;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.storm.kafka.spout.internal.OffsetManager;
+import org.apache.storm.metric.api.IMetric;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * This class is used compute the partition and topic level offset metrics
+ * <p>
+ * Partition level metrics are:
+ * topicName/partition_{number}/earliestTimeOffset //gives beginning offset of the partition
+ * topicName/partition_{number}/latestTimeOffset //gives end offset of the partition
+ * topicName/partition_{number}/latestEmittedOffset //gives latest emitted offset of the partition from the spout
+ * topicName/partition_{number}/latestCompletedOffset //gives latest committed offset of the partition from the spout
+ * topicName/partition_{number}/spoutLag // the delta between the latest Offset and latestCompletedOffset
+ * topicName/partition_{number}/recordsInPartition // total number of records in the partition
+ * </p>
+ * <p>
+ * Topic level metrics are:
+ * topicName/totalEarliestTimeOffset //gives the total beginning offset of all the associated partitions of this spout
+ * topicName/totalLatestTimeOffset //gives the total end offset of all the associated partitions of this spout
+ * topicName/totalLatestEmittedOffset //gives the total latest emitted offset of all the associated partitions of this spout
+ * topicName/totalLatestCompletedOffset //gives the total latest committed offset of all the associated partitions of this spout
+ * topicName/spoutLag // total spout lag of all the associated partitions of this spout
+ * topicName/totalRecordsInPartitions //total number of records in all the associated partitions of this spout
+ * </p>
+ */
+public class KafkaOffsetMetric implements IMetric {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaOffsetMetric.class);
+ private final Supplier<Map<TopicPartition, OffsetManager>> offsetManagerSupplier;
+ private final Supplier<KafkaConsumer> consumerSupplier;
+
+ public KafkaOffsetMetric(Supplier offsetManagerSupplier, Supplier consumerSupplier) {
+ this.offsetManagerSupplier = offsetManagerSupplier;
+ this.consumerSupplier = consumerSupplier;
+ }
+
+ @Override
+ public Object getValueAndReset() {
+
+ Map<TopicPartition, OffsetManager> offsetManagers = offsetManagerSupplier.get();
+ KafkaConsumer kafkaConsumer = consumerSupplier.get();
+
+ if (offsetManagers == null || offsetManagers.isEmpty() || kafkaConsumer == null) {
+ LOG.debug("Metrics Tick: offsetManagers or kafkaConsumer is null.");
+ return null;
+ }
+
+ Map<String,TopicMetrics> topicMetricsMap = new HashMap<>();
+ Set<TopicPartition> topicPartitions = offsetManagers.keySet();
+
+ Map<TopicPartition, Long> beginningOffsets = kafkaConsumer.beginningOffsets(topicPartitions);
+ Map<TopicPartition, Long> endOffsets = kafkaConsumer.endOffsets(topicPartitions);
+ //map to hold partition level and topic level metrics
+ Map<String, Long> result = new HashMap<>();
+
+ for (Map.Entry<TopicPartition, OffsetManager> entry : offsetManagers.entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ OffsetManager offsetManager = entry.getValue();
+
+ long latestTimeOffset = endOffsets.get(topicPartition);
+ long earliestTimeOffset = beginningOffsets.get(topicPartition);
+
+ long latestEmittedOffset = offsetManager.getLatestEmittedOffset();
+ long latestCompletedOffset = offsetManager.getCommittedOffset();
+ long spoutLag = latestTimeOffset - latestCompletedOffset;
+ long recordsInPartition = latestTimeOffset - earliestTimeOffset;
+
+ String metricPath = topicPartition.topic() + "/partition_" + topicPartition.partition();
+ result.put(metricPath + "/" + "spoutLag", spoutLag);
+ result.put(metricPath + "/" + "earliestTimeOffset", earliestTimeOffset);
+ result.put(metricPath + "/" + "latestTimeOffset", latestTimeOffset);
+ result.put(metricPath + "/" + "latestEmittedOffset", latestEmittedOffset);
+ result.put(metricPath + "/" + "latestCompletedOffset", latestCompletedOffset);
+ result.put(metricPath + "/" + "recordsInPartition", recordsInPartition);
+
+ TopicMetrics topicMetrics = topicMetricsMap.get(topicPartition.topic());
+ if (topicMetrics == null) {
+ topicMetrics = new TopicMetrics();
+ topicMetricsMap.put(topicPartition.topic(), topicMetrics);
+ }
+
+ topicMetrics.totalSpoutLag += spoutLag;
+ topicMetrics.totalEarliestTimeOffset += earliestTimeOffset;
+ topicMetrics.totalLatestTimeOffset += latestTimeOffset;
+ topicMetrics.totalLatestEmittedOffset += latestEmittedOffset;
+ topicMetrics.totalLatestCompletedOffset += latestCompletedOffset;
+ topicMetrics.totalRecordsInPartitions += recordsInPartition;
+ }
+
+ for (Map.Entry<String, TopicMetrics> e : topicMetricsMap.entrySet()) {
+ String topic = e.getKey();
+ TopicMetrics topicMetrics = e.getValue();
+ result.put(topic + "/" + "totalSpoutLag", topicMetrics.totalSpoutLag);
+ result.put(topic + "/" + "totalEarliestTimeOffset", topicMetrics.totalEarliestTimeOffset);
+ result.put(topic + "/" + "totalLatestTimeOffset", topicMetrics.totalLatestTimeOffset);
+ result.put(topic + "/" + "totalLatestEmittedOffset", topicMetrics.totalLatestEmittedOffset);
+ result.put(topic + "/" + "totalLatestCompletedOffset", topicMetrics.totalLatestCompletedOffset);
+ result.put(topic + "/" + "totalRecordsInPartitions", topicMetrics.totalRecordsInPartitions);
+ }
+
+ LOG.debug("Metrics Tick: value : {}", result);
+ return result;
+ }
+
+ private class TopicMetrics {
+ long totalSpoutLag = 0;
+ long totalEarliestTimeOffset = 0;
+ long totalLatestTimeOffset = 0;
+ long totalLatestEmittedOffset = 0;
+ long totalLatestCompletedOffset = 0;
+ long totalRecordsInPartitions = 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutBatchMetadata.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutBatchMetadata.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutBatchMetadata.java
index 7079cb2..6fa81aa 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutBatchMetadata.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutBatchMetadata.java
@@ -18,6 +18,11 @@
package org.apache.storm.kafka.spout.trident;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang.Validate;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
@@ -30,37 +35,47 @@ import java.util.List;
/**
* Wraps transaction batch information
*/
-public class KafkaTridentSpoutBatchMetadata<K,V> implements Serializable {
+public class KafkaTridentSpoutBatchMetadata implements Serializable {
private static final Logger LOG = LoggerFactory.getLogger(KafkaTridentSpoutBatchMetadata.class);
-
+ private static final TopicPartitionSerializer TP_SERIALIZER = new TopicPartitionSerializer();
+
+ public static final String TOPIC_PARTITION_KEY = "tp";
+ public static final String FIRST_OFFSET_KEY = "firstOffset";
+ public static final String LAST_OFFSET_KEY = "lastOffset";
+
// topic partition of this batch
- private TopicPartition topicPartition;
+ private final TopicPartition topicPartition;
// first offset of this batch
- private long firstOffset;
+ private final long firstOffset;
// last offset of this batch
- private long lastOffset;
+ private final long lastOffset;
+ /**
+ * Builds a metadata object.
+ * @param topicPartition The topic partition
+ * @param firstOffset The first offset for the batch
+ * @param lastOffset The last offset for the batch
+ */
public KafkaTridentSpoutBatchMetadata(TopicPartition topicPartition, long firstOffset, long lastOffset) {
this.topicPartition = topicPartition;
this.firstOffset = firstOffset;
this.lastOffset = lastOffset;
}
- public KafkaTridentSpoutBatchMetadata(TopicPartition topicPartition, ConsumerRecords<K, V> consumerRecords, KafkaTridentSpoutBatchMetadata<K, V> lastBatch) {
- this.topicPartition = topicPartition;
-
+ /**
+ * Builds a metadata object from a non-empty set of records.
+ * @param topicPartition The topic partition the records belong to.
+ * @param consumerRecords The non-empty set of records.
+ */
+ public <K, V> KafkaTridentSpoutBatchMetadata(TopicPartition topicPartition, ConsumerRecords<K, V> consumerRecords) {
+ Validate.notNull(consumerRecords.records(topicPartition));
List<ConsumerRecord<K, V>> records = consumerRecords.records(topicPartition);
-
- if (records != null && !records.isEmpty()) {
- firstOffset = records.get(0).offset();
- lastOffset = records.get(records.size() - 1).offset();
- } else {
- if (lastBatch != null) {
- firstOffset = lastBatch.firstOffset;
- lastOffset = lastBatch.lastOffset;
- }
- }
- LOG.debug("Created {}", this);
+ Validate.isTrue(!records.isEmpty(), "There must be at least one record in order to build metadata");
+
+ this.topicPartition = topicPartition;
+ firstOffset = records.get(0).offset();
+ lastOffset = records.get(records.size() - 1).offset();
+ LOG.debug("Created {}", this.toString());
}
public long getFirstOffset() {
@@ -74,9 +89,32 @@ public class KafkaTridentSpoutBatchMetadata<K,V> implements Serializable {
public TopicPartition getTopicPartition() {
return topicPartition;
}
+
+ /**
+ * Constructs a metadata object from a Map in the format produced by {@link #toMap() }.
+ * @param map The source map
+ * @return A new metadata object
+ */
+ public static KafkaTridentSpoutBatchMetadata fromMap(Map<String, Object> map) {
+ Map<String, Object> topicPartitionMap = (Map<String, Object>)map.get(TOPIC_PARTITION_KEY);
+ TopicPartition tp = TP_SERIALIZER.fromMap(topicPartitionMap);
+ return new KafkaTridentSpoutBatchMetadata(tp, ((Number)map.get(FIRST_OFFSET_KEY)).longValue(),
+ ((Number)map.get(LAST_OFFSET_KEY)).longValue());
+ }
+
+ /**
+ * Writes this metadata object to a Map so Trident can read/write it to Zookeeper.
+ */
+ public Map<String, Object> toMap() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(TOPIC_PARTITION_KEY, TP_SERIALIZER.toMap(topicPartition));
+ map.put(FIRST_OFFSET_KEY, firstOffset);
+ map.put(LAST_OFFSET_KEY, lastOffset);
+ return map;
+ }
@Override
- public String toString() {
+ public final String toString() {
return super.toString() +
"{topicPartition=" + topicPartition +
", firstOffset=" + firstOffset +
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
index 87af0cf..3b4aa4b 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
@@ -18,10 +18,11 @@
package org.apache.storm.kafka.spout.trident;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.RecordTranslator;
@@ -48,9 +49,9 @@ import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrat
import static org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_LATEST;
public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTridentSpout.Emitter<
- List<TopicPartition>,
+ List<Map<String, Object>>,
KafkaTridentSpoutTopicPartition,
- KafkaTridentSpoutBatchMetadata<K, V>>,
+ Map<String, Object>>,
Serializable {
private static final long serialVersionUID = -7343927794834130435L;
@@ -61,16 +62,24 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
// Bookkeeping
private final KafkaTridentSpoutManager<K, V> kafkaManager;
- private Set<TopicPartition> firstPoll = new HashSet<>(); // set of topic-partitions for which first poll has already occurred
+ // set of topic-partitions for which first poll has already occurred, and the first polled txid
+ private final Map<TopicPartition, Long> firstPollTransaction = new HashMap<>();
// Declare some KafkaTridentSpoutManager references for convenience
private final long pollTimeoutMs;
private final KafkaSpoutConfig.FirstPollOffsetStrategy firstPollOffsetStrategy;
private final RecordTranslator<K, V> translator;
private final Timer refreshSubscriptionTimer;
+ private final TopicPartitionSerializer tpSerializer = new TopicPartitionSerializer();
private TopologyContext topologyContext;
+ /**
+ * Create a new Kafka spout emitter.
+ * @param kafkaManager The Kafka consumer manager to use
+ * @param topologyContext The topology context
+ * @param refreshSubscriptionTimer The timer for deciding when to recheck the subscription
+ */
public KafkaTridentSpoutEmitter(KafkaTridentSpoutManager<K, V> kafkaManager, TopologyContext topologyContext, Timer refreshSubscriptionTimer) {
this.kafkaConsumer = kafkaManager.createAndSubscribeKafkaConsumer(topologyContext);
this.kafkaManager = kafkaManager;
@@ -81,7 +90,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
final KafkaSpoutConfig<K, V> kafkaSpoutConfig = kafkaManager.getKafkaSpoutConfig();
this.pollTimeoutMs = kafkaSpoutConfig.getPollTimeoutMs();
this.firstPollOffsetStrategy = kafkaSpoutConfig.getFirstPollOffsetStrategy();
- LOG.debug("Created {}", this);
+ LOG.debug("Created {}", this.toString());
}
/**
@@ -93,15 +102,16 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
}
@Override
- public KafkaTridentSpoutBatchMetadata<K, V> emitPartitionBatch(TransactionAttempt tx, TridentCollector collector,
- KafkaTridentSpoutTopicPartition currBatchPartition, KafkaTridentSpoutBatchMetadata<K, V> lastBatch) {
+ public Map<String, Object> emitPartitionBatch(TransactionAttempt tx, TridentCollector collector,
+ KafkaTridentSpoutTopicPartition currBatchPartition, Map<String, Object> lastBatch) {
LOG.debug("Processing batch: [transaction = {}], [currBatchPartition = {}], [lastBatchMetadata = {}], [collector = {}]",
tx, currBatchPartition, lastBatch, collector);
final TopicPartition currBatchTp = currBatchPartition.getTopicPartition();
final Set<TopicPartition> assignments = kafkaConsumer.assignment();
- KafkaTridentSpoutBatchMetadata<K, V> currentBatch = lastBatch;
+ KafkaTridentSpoutBatchMetadata lastBatchMeta = lastBatch == null ? null : KafkaTridentSpoutBatchMetadata.fromMap(lastBatch);
+ KafkaTridentSpoutBatchMetadata currentBatch = lastBatchMeta;
Collection<TopicPartition> pausedTopicPartitions = Collections.emptySet();
if (assignments == null || !assignments.contains(currBatchPartition.getTopicPartition())) {
@@ -114,7 +124,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
// pause other topic-partitions to only poll from current topic-partition
pausedTopicPartitions = pauseTopicPartitions(currBatchTp);
- seek(currBatchTp, lastBatch);
+ seek(currBatchTp, lastBatchMeta, tx.getTransactionId());
// poll
if (refreshSubscriptionTimer.isExpiredResetOnTrue()) {
@@ -127,7 +137,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
if (!records.isEmpty()) {
emitTuples(collector, records);
// build new metadata
- currentBatch = new KafkaTridentSpoutBatchMetadata<>(currBatchTp, records, lastBatch);
+ currentBatch = new KafkaTridentSpoutBatchMetadata(currBatchTp, records);
}
} finally {
kafkaConsumer.resume(pausedTopicPartitions);
@@ -137,7 +147,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
"[currBatchMetadata = {}], [collector = {}]", tx, currBatchPartition, lastBatch, currentBatch, collector);
}
- return currentBatch;
+ return currentBatch == null ? null : currentBatch.toMap();
}
private void emitTuples(TridentCollector collector, ConsumerRecords<K, V> records) {
@@ -149,46 +159,49 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
}
/**
- * Determines the offset of the next fetch. For failed batches lastBatchMeta is not null and contains the fetch
- * offset of the failed batch. In this scenario the next fetch will take place at offset of the failed batch + 1.
- * When the previous batch is successful, lastBatchMeta is null, and the offset of the next fetch is either the
- * offset of the last commit to kafka, or if no commit was yet made, the offset dictated by
- * {@link KafkaSpoutConfig.FirstPollOffsetStrategy}
+ * Determines the offset of the next fetch. Will use the firstPollOffsetStrategy if this is the first poll for the topic partition.
+ * Otherwise the next offset will be one past the last batch, based on lastBatchMeta.
+ *
+ * <p>lastBatchMeta should only be null when the previous txid was not emitted (e.g. new topic),
+ * it is the first poll for the spout instance, or it is a replay of the first txid this spout emitted on this partition.
+ * In the second case, there are either no previous transactions, or the MBC is still committing them
+ * and they will fail because this spout did not emit the corresponding batches. If it had emitted them, the meta could not be null.
+ * In any case, the lastBatchMeta should never be null if this is not the first poll for this spout instance.
*
* @return the offset of the next fetch
*/
- private long seek(TopicPartition tp, KafkaTridentSpoutBatchMetadata<K, V> lastBatchMeta) {
- if (lastBatchMeta != null) {
- kafkaConsumer.seek(tp, lastBatchMeta.getLastOffset() + 1); // seek next offset after last offset from previous batch
- LOG.debug("Seeking fetch offset to next offset after last offset from previous batch for topic-partition [{}]", tp);
- } else if (isFirstPoll(tp)) {
- LOG.debug("Seeking fetch offset from firstPollOffsetStrategy and last commit to Kafka for topic-partition [{}]", tp);
- firstPoll.add(tp);
- final OffsetAndMetadata committedOffset = kafkaConsumer.committed(tp);
- if (committedOffset != null) { // offset was committed for this TopicPartition
- if (firstPollOffsetStrategy.equals(EARLIEST)) {
- kafkaConsumer.seekToBeginning(Collections.singleton(tp));
- } else if (firstPollOffsetStrategy.equals(LATEST)) {
- kafkaConsumer.seekToEnd(Collections.singleton(tp));
- } else {
- // By default polling starts at the last committed offset. +1 to point fetch to the first uncommitted offset.
- kafkaConsumer.seek(tp, committedOffset.offset() + 1);
- }
- } else { // no commits have ever been done, so start at the beginning or end depending on the strategy
- if (firstPollOffsetStrategy.equals(EARLIEST) || firstPollOffsetStrategy.equals(UNCOMMITTED_EARLIEST)) {
- kafkaConsumer.seekToBeginning(Collections.singleton(tp));
- } else if (firstPollOffsetStrategy.equals(LATEST) || firstPollOffsetStrategy.equals(UNCOMMITTED_LATEST)) {
- kafkaConsumer.seekToEnd(Collections.singleton(tp));
- }
+ private long seek(TopicPartition tp, KafkaTridentSpoutBatchMetadata lastBatchMeta, long transactionId) {
+ if (isFirstPoll(tp, transactionId)) {
+ if (firstPollOffsetStrategy == EARLIEST) {
+ LOG.debug("First poll for topic partition [{}], seeking to partition beginning", tp);
+ kafkaConsumer.seekToBeginning(Collections.singleton(tp));
+ } else if (firstPollOffsetStrategy == LATEST) {
+ LOG.debug("First poll for topic partition [{}], seeking to partition end", tp);
+ kafkaConsumer.seekToEnd(Collections.singleton(tp));
+ } else if (lastBatchMeta != null) {
+ LOG.debug("First poll for topic partition [{}], using last batch metadata", tp);
+ kafkaConsumer.seek(tp, lastBatchMeta.getLastOffset() + 1); // seek next offset after last offset from previous batch
+ } else if (firstPollOffsetStrategy == UNCOMMITTED_EARLIEST) {
+ LOG.debug("First poll for topic partition [{}] with no last batch metadata, seeking to partition beginning", tp);
+ kafkaConsumer.seekToBeginning(Collections.singleton(tp));
+ } else if (firstPollOffsetStrategy == UNCOMMITTED_LATEST) {
+ LOG.debug("First poll for topic partition [{}] with no last batch metadata, seeking to partition end", tp);
+ kafkaConsumer.seekToEnd(Collections.singleton(tp));
}
+ firstPollTransaction.put(tp, transactionId);
+ } else {
+ kafkaConsumer.seek(tp, lastBatchMeta.getLastOffset() + 1); // seek next offset after last offset from previous batch
+ LOG.debug("First poll for topic partition [{}], using last batch metadata", tp);
}
+
final long fetchOffset = kafkaConsumer.position(tp);
LOG.debug("Set [fetchOffset = {}]", fetchOffset);
return fetchOffset;
}
- private boolean isFirstPoll(TopicPartition tp) {
- return !firstPoll.contains(tp);
+ private boolean isFirstPoll(TopicPartition tp, long txid) {
+ // The first poll is either the "real" first transaction, or a replay of the first transaction
+ return !firstPollTransaction.containsKey(tp) || firstPollTransaction.get(tp) == txid;
}
// returns paused topic-partitions.
@@ -215,15 +228,20 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
* @return ordered list of topic partitions for this task
*/
@Override
- public List<KafkaTridentSpoutTopicPartition> getOrderedPartitions(final List<TopicPartition> allPartitionInfo) {
- final List<KafkaTridentSpoutTopicPartition> allPartitions = newKafkaTridentSpoutTopicPartitions(allPartitionInfo);
+ public List<KafkaTridentSpoutTopicPartition> getOrderedPartitions(final List<Map<String, Object>> allPartitionInfo) {
+ List<TopicPartition> allTopicPartitions = new ArrayList<>();
+ for(Map<String, Object> map : allPartitionInfo) {
+ allTopicPartitions.add(tpSerializer.fromMap(map));
+ }
+ final List<KafkaTridentSpoutTopicPartition> allPartitions = newKafkaTridentSpoutTopicPartitions(allTopicPartitions);
LOG.debug("Returning all topic-partitions {} across all tasks. Current task index [{}]. Total tasks [{}] ",
allPartitions, topologyContext.getThisTaskIndex(), getNumTasks());
return allPartitions;
}
@Override
- public List<KafkaTridentSpoutTopicPartition> getPartitionsForTask(int taskId, int numTasks, List<TopicPartition> allPartitionInfo) {
+ public List<KafkaTridentSpoutTopicPartition> getPartitionsForTask(int taskId, int numTasks,
+ List<Map<String, Object>> allPartitionInfo) {
final Set<TopicPartition> assignedTps = kafkaConsumer.assignment();
LOG.debug("Consumer [{}], running on task with index [{}], has assigned topic-partitions {}", kafkaConsumer, taskId, assignedTps);
final List<KafkaTridentSpoutTopicPartition> taskTps = newKafkaTridentSpoutTopicPartitions(assignedTps);
@@ -253,7 +271,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
}
@Override
- public String toString() {
+ public final String toString() {
return super.toString() +
"{kafkaManager=" + kafkaManager +
'}';
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java
index 4054b49..26db5c9 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutManager.java
@@ -43,15 +43,18 @@ public class KafkaTridentSpoutManager<K, V> implements Serializable {
// Declare some KafkaSpoutConfig references for convenience
private Fields fields;
+ /**
+ * Create a KafkaConsumer manager for the trident spout.
+ * @param kafkaSpoutConfig The consumer config
+ */
public KafkaTridentSpoutManager(KafkaSpoutConfig<K, V> kafkaSpoutConfig) {
this.kafkaSpoutConfig = kafkaSpoutConfig;
this.fields = getFields();
- LOG.debug("Created {}", this);
+ LOG.debug("Created {}", this.toString());
}
KafkaConsumer<K,V> createAndSubscribeKafkaConsumer(TopologyContext context) {
- kafkaConsumer = new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps(),
- kafkaSpoutConfig.getKeyDeserializer(), kafkaSpoutConfig.getValueDeserializer());
+ kafkaConsumer = new KafkaConsumer<>(kafkaSpoutConfig.getKafkaProps());
kafkaSpoutConfig.getSubscription().subscribe(kafkaConsumer, new KafkaSpoutConsumerRebalanceListener(), context);
return kafkaConsumer;
@@ -65,7 +68,7 @@ public class KafkaTridentSpoutManager<K, V> implements Serializable {
return KafkaTridentSpoutTopicPartitionRegistry.INSTANCE.getTopicPartitions();
}
- Fields getFields() {
+ final Fields getFields() {
if (fields == null) {
RecordTranslator<K, V> translator = kafkaSpoutConfig.getTranslator();
Fields fs = null;
@@ -89,7 +92,7 @@ public class KafkaTridentSpoutManager<K, V> implements Serializable {
}
@Override
- public String toString() {
+ public final String toString() {
return super.toString() +
"{kafkaConsumer=" + kafkaConsumer +
", kafkaSpoutConfig=" + kafkaSpoutConfig +
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
index 0f7f0af..ecc9219 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
@@ -18,7 +18,8 @@
package org.apache.storm.kafka.spout.trident;
-import org.apache.kafka.common.TopicPartition;
+import java.util.List;
+import java.util.Map;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
@@ -26,10 +27,8 @@ import org.apache.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
-import java.util.Map;
-
-public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSpout<List<TopicPartition>, KafkaTridentSpoutTopicPartition, KafkaTridentSpoutBatchMetadata<K,V>> {
+public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSpout<List<Map<String, Object>>,
+ KafkaTridentSpoutTopicPartition, Map<String, Object>> {
private static final long serialVersionUID = -8003272486566259640L;
private static final Logger LOG = LoggerFactory.getLogger(KafkaTridentSpoutOpaque.class);
@@ -42,16 +41,17 @@ public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSp
public KafkaTridentSpoutOpaque(KafkaTridentSpoutManager<K, V> kafkaManager) {
this.kafkaManager = kafkaManager;
- LOG.debug("Created {}", this);
+ LOG.debug("Created {}", this.toString());
}
@Override
- public Emitter<List<TopicPartition>, KafkaTridentSpoutTopicPartition, KafkaTridentSpoutBatchMetadata<K,V>> getEmitter(Map conf, TopologyContext context) {
+ public Emitter<List<Map<String, Object>>, KafkaTridentSpoutTopicPartition, Map<String, Object>> getEmitter(
+ Map conf, TopologyContext context) {
return new KafkaTridentSpoutEmitter<>(kafkaManager, context);
}
@Override
- public Coordinator<List<TopicPartition>> getCoordinator(Map conf, TopologyContext context) {
+ public Coordinator<List<Map<String, Object>>> getCoordinator(Map conf, TopologyContext context) {
return new KafkaTridentSpoutOpaqueCoordinator<>(kafkaManager);
}
@@ -68,7 +68,7 @@ public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSp
}
@Override
- public String toString() {
+ public final String toString() {
return super.toString() +
"{kafkaManager=" + kafkaManager + '}';
}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java
index 7898b6e..449e24b 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java
@@ -18,23 +18,25 @@
package org.apache.storm.kafka.spout.trident;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
import org.apache.kafka.common.TopicPartition;
import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-public class KafkaTridentSpoutOpaqueCoordinator<K,V> implements IOpaquePartitionedTridentSpout.Coordinator<List<TopicPartition>>, Serializable {
+public class KafkaTridentSpoutOpaqueCoordinator<K,V> implements IOpaquePartitionedTridentSpout.Coordinator<List<Map<String, Object>>>,
+ Serializable {
private static final Logger LOG = LoggerFactory.getLogger(KafkaTridentSpoutOpaqueCoordinator.class);
- private KafkaTridentSpoutManager<K,V> kafkaManager;
+ private final TopicPartitionSerializer tpSerializer = new TopicPartitionSerializer();
+ private final KafkaTridentSpoutManager<K,V> kafkaManager;
public KafkaTridentSpoutOpaqueCoordinator(KafkaTridentSpoutManager<K, V> kafkaManager) {
this.kafkaManager = kafkaManager;
- LOG.debug("Created {}", this);
+ LOG.debug("Created {}", this.toString());
}
@Override
@@ -44,10 +46,14 @@ public class KafkaTridentSpoutOpaqueCoordinator<K,V> implements IOpaquePartition
}
@Override
- public List<TopicPartition> getPartitionsForBatch() {
+ public List<Map<String, Object>> getPartitionsForBatch() {
final ArrayList<TopicPartition> topicPartitions = new ArrayList<>(kafkaManager.getTopicPartitions());
LOG.debug("TopicPartitions for batch {}", topicPartitions);
- return topicPartitions;
+ List<Map<String, Object>> tps = new ArrayList<>();
+ for(TopicPartition tp : topicPartitions) {
+ tps.add(tpSerializer.toMap(tp));
+ }
+ return tps;
}
@Override
@@ -56,7 +62,7 @@ public class KafkaTridentSpoutOpaqueCoordinator<K,V> implements IOpaquePartition
}
@Override
- public String toString() {
+ public final String toString() {
return super.toString() +
"{kafkaManager=" + kafkaManager +
'}';
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/TopicPartitionSerializer.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/TopicPartitionSerializer.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/TopicPartitionSerializer.java
new file mode 100644
index 0000000..50e78f0
--- /dev/null
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/TopicPartitionSerializer.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2017 The Apache Software Foundation.
+ *
+ * Licensed 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.kafka.spout.trident;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.TopicPartition;
+
+public class TopicPartitionSerializer {
+
+ public static final String TOPIC_PARTITION_TOPIC_KEY = "topic";
+ public static final String TOPIC_PARTITION_PARTITION_KEY = "partition";
+
+ /**
+ * Serializes the given TopicPartition to Map so Trident can serialize it to JSON.
+ */
+ public Map<String, Object> toMap(TopicPartition topicPartition) {
+ Map<String, Object> topicPartitionMap = new HashMap<>();
+ topicPartitionMap.put(TOPIC_PARTITION_TOPIC_KEY, topicPartition.topic());
+ topicPartitionMap.put(TOPIC_PARTITION_PARTITION_KEY, topicPartition.partition());
+ return topicPartitionMap;
+ }
+
+ /**
+ * Deserializes the given map into a TopicPartition. The map keys are expected to be those produced by
+ * {@link #toMap(org.apache.kafka.common.TopicPartition)}.
+ */
+ public TopicPartition fromMap(Map<String, Object> map) {
+ return new TopicPartition((String) map.get(TOPIC_PARTITION_TOPIC_KEY),
+ ((Number) map.get(TOPIC_PARTITION_PARTITION_KEY)).intValue());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/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 76e0f02..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;
}
@@ -66,36 +65,45 @@ public class TridentKafkaState implements State {
LOG.debug("commit is Noop.");
}
+ /**
+ * Prepare this State.
+ * @param options The KafkaProducer config.
+ */
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);
}
+ /**
+ * Write the given tuples to Kafka.
+ * @param tuples The tuples to write.
+ * @param collector The Trident collector.
+ */
public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
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) {
@@ -106,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/324bc95b/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/324bc95b/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);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java
new file mode 100644
index 0000000..0467383
--- /dev/null
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutAbstractTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.kafka.spout;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.storm.kafka.KafkaUnitRule;
+import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration;
+import org.apache.storm.kafka.spout.internal.KafkaConsumerFactory;
+import org.apache.storm.kafka.spout.internal.KafkaConsumerFactoryDefault;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Time;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.MockitoAnnotations;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+public abstract class KafkaSpoutAbstractTest {
+ @Rule
+ public KafkaUnitRule kafkaUnitRule = new KafkaUnitRule();
+
+ final TopologyContext topologyContext = mock(TopologyContext.class);
+ final Map<String, Object> conf = new HashMap<>();
+ final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class);
+ final long commitOffsetPeriodMs;
+
+ KafkaConsumer<String, String> consumerSpy;
+ KafkaSpout<String, String> spout;
+
+ @Captor
+ ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture;
+ private Time.SimulatedTime simulatedTime;
+ private KafkaSpoutConfig<String, String> spoutConfig;
+
+ /**
+ * This constructor should be called by the subclass' default constructor with the desired value
+ * @param commitOffsetPeriodMs commit offset period to be used in commit and verification of messages committed
+ */
+ protected KafkaSpoutAbstractTest(long commitOffsetPeriodMs) {
+ this.commitOffsetPeriodMs = commitOffsetPeriodMs;
+ }
+
+ @Before
+ public void setUp() {
+ MockitoAnnotations.initMocks(this);
+
+ spoutConfig = createSpoutConfig();
+
+ consumerSpy = createConsumerSpy();
+
+ spout = new KafkaSpout<>(spoutConfig, createConsumerFactory());
+
+ simulatedTime = new Time.SimulatedTime();
+ }
+
+ private KafkaConsumerFactory<String, String> createConsumerFactory() {
+
+ return new KafkaConsumerFactory<String, String>() {
+ @Override
+ public KafkaConsumer<String, String> createConsumer(KafkaSpoutConfig<String, String> kafkaSpoutConfig) {
+ return consumerSpy;
+ }
+
+ };
+ }
+
+ KafkaConsumer<String, String> createConsumerSpy() {
+ return spy(new KafkaConsumerFactoryDefault<String, String>().createConsumer(spoutConfig));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ simulatedTime.close();
+ }
+
+ abstract KafkaSpoutConfig<String, String> createSpoutConfig();
+
+ void prepareSpout(int messageCount) throws Exception {
+ SingleTopicKafkaUnitSetupHelper.populateTopicData(kafkaUnitRule.getKafkaUnit(), SingleTopicKafkaSpoutConfiguration.TOPIC, messageCount);
+ SingleTopicKafkaUnitSetupHelper.initializeSpout(spout, conf, topologyContext, collectorMock);
+ }
+
+ /**
+ * Helper method to in sequence do:
+ * <li>
+ * <ul>spout.nexTuple()</ul>
+ * <ul>verify messageId</ul>
+ * <ul>spout.ack(msgId)</ul>
+ * <ul>reset(collector) to be able to reuse mock</ul>
+ * </li>
+ *
+ * @param offset offset of message to be verified
+ * @return {@link ArgumentCaptor} of the messageId verified
+ */
+ ArgumentCaptor<Object> nextTuple_verifyEmitted_ack_resetCollector(int offset) {
+ spout.nextTuple();
+
+ ArgumentCaptor<Object> messageId = verifyMessageEmitted(offset);
+
+ spout.ack(messageId.getValue());
+
+ reset(collectorMock);
+
+ return messageId;
+ }
+
+ // offset and messageId are used interchangeably
+ ArgumentCaptor<Object> verifyMessageEmitted(int offset) {
+ final ArgumentCaptor<Object> messageId = ArgumentCaptor.forClass(Object.class);
+
+ verify(collectorMock).emit(
+ eq(SingleTopicKafkaSpoutConfiguration.STREAM),
+ eq(new Values(SingleTopicKafkaSpoutConfiguration.TOPIC,
+ Integer.toString(offset),
+ Integer.toString(offset))),
+ messageId.capture());
+
+ return messageId;
+ }
+
+ void commitAndVerifyAllMessagesCommitted(long msgCount) {
+ // reset commit timer such that commit happens on next call to nextTuple()
+ Time.advanceTime(commitOffsetPeriodMs + KafkaSpout.TIMER_DELAY_MS);
+
+ //Commit offsets
+ spout.nextTuple();
+
+ verifyAllMessagesCommitted(msgCount);
+ }
+
+ /*
+ * Asserts that commitSync has been called once,
+ * that there are only commits on one topic,
+ * and that the committed offset covers messageCount messages
+ */
+ void verifyAllMessagesCommitted(long messageCount) {
+ verify(consumerSpy).commitSync(commitCapture.capture());
+
+ final Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue();
+ assertThat("Expected commits for only one topic partition", commits.entrySet().size(), is(1));
+
+ OffsetAndMetadata offset = commits.entrySet().iterator().next().getValue();
+ assertThat("Expected committed offset to cover all emitted messages", offset.offset(), is(messageCount));
+
+ reset(consumerSpy);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutCommitTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutCommitTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutCommitTest.java
deleted file mode 100644
index 3981a9a..0000000
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutCommitTest.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * Copyright 2017 The Apache Software Foundation.
- *
- * Licensed 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.kafka.spout;
-
-import static org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration.getKafkaSpoutConfigBuilder;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.kafka.clients.consumer.*;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.storm.kafka.spout.builders.SingleTopicKafkaSpoutConfiguration;
-import org.apache.storm.kafka.spout.internal.KafkaConsumerFactory;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.utils.Time;
-import org.apache.storm.utils.Time.SimulatedTime;
-import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
-import org.mockito.InOrder;
-import org.mockito.MockitoAnnotations;
-
-public class KafkaSpoutCommitTest {
-
- private final long offsetCommitPeriodMs = 2_000;
- private final TopologyContext contextMock = mock(TopologyContext.class);
- private final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class);
- private final Map<String, Object> conf = new HashMap<>();
- private final TopicPartition partition = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 1);
- private KafkaConsumer<String, String> consumerMock;
- private KafkaSpout<String, String> spout;
- private KafkaSpoutConfig spoutConfig;
-
- @Captor
- private ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>> commitCapture;
-
- private void setupSpout(Set<TopicPartition> assignedPartitions) {
- MockitoAnnotations.initMocks(this);
- spoutConfig = getKafkaSpoutConfigBuilder(-1)
- .setOffsetCommitPeriodMs(offsetCommitPeriodMs)
- .build();
-
- consumerMock = mock(KafkaConsumer.class);
- KafkaConsumerFactory<String, String> consumerFactory = new KafkaConsumerFactory<String, String>() {
- @Override
- public KafkaConsumer<String, String> createConsumer(KafkaSpoutConfig<String, String> kafkaSpoutConfig) {
- return consumerMock;
- }
- };
-
- //Set up a spout listening to 1 topic partition
- spout = new KafkaSpout<>(spoutConfig, consumerFactory);
-
- spout.open(conf, contextMock, collectorMock);
- spout.activate();
-
- ArgumentCaptor<ConsumerRebalanceListener> rebalanceListenerCapture = ArgumentCaptor.forClass(ConsumerRebalanceListener.class);
- verify(consumerMock).subscribe(anyCollection(), rebalanceListenerCapture.capture());
-
- //Assign partitions to the spout
- ConsumerRebalanceListener consumerRebalanceListener = rebalanceListenerCapture.getValue();
- consumerRebalanceListener.onPartitionsAssigned(assignedPartitions);
- }
-
- @Test
- public void testCommitSuccessWithOffsetVoids() {
- //Verify that the commit logic can handle offset voids
- try (SimulatedTime simulatedTime = new SimulatedTime()) {
- setupSpout(Collections.singleton(partition));
- Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>();
- List<ConsumerRecord<String, String>> recordsForPartition = new ArrayList<>();
- // Offsets emitted are 0,1,2,3,4,<void>,8,9
- for (int i = 0; i < 5; i++) {
- recordsForPartition.add(new ConsumerRecord(partition.topic(), partition.partition(), i, "key", "value"));
- }
- for (int i = 8; i < 10; i++) {
- recordsForPartition.add(new ConsumerRecord(partition.topic(), partition.partition(), i, "key", "value"));
- }
- records.put(partition, recordsForPartition);
-
- when(consumerMock.poll(anyLong()))
- .thenReturn(new ConsumerRecords(records));
-
- for (int i = 0; i < recordsForPartition.size(); i++) {
- spout.nextTuple();
- }
-
- ArgumentCaptor<KafkaSpoutMessageId> messageIds = ArgumentCaptor.forClass(KafkaSpoutMessageId.class);
- verify(collectorMock, times(recordsForPartition.size())).emit(anyString(), anyList(), messageIds.capture());
-
- for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) {
- spout.ack(messageId);
- }
-
- // Advance time and then trigger first call to kafka consumer commit; the commit must progress to offset 9
- Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs);
- Map<TopicPartition, List<ConsumerRecord<String, String>>> emptyConsumerRecords = Collections.emptyMap();
- when(consumerMock.poll(anyLong()))
- .thenReturn(new ConsumerRecords<>(emptyConsumerRecords));
- spout.nextTuple();
-
- InOrder inOrder = inOrder(consumerMock);
- inOrder.verify(consumerMock).commitSync(commitCapture.capture());
- inOrder.verify(consumerMock).poll(anyLong());
-
- //verify that Offset 10 was last committed offset, since this is the offset the spout should resume at
- Map<TopicPartition, OffsetAndMetadata> commits = commitCapture.getValue();
- assertTrue(commits.containsKey(partition));
- assertEquals(10, commits.get(partition).offset());
- }
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/324bc95b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java
index 9f62b90..90e906b 100644
--- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java
+++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutConfigTest.java
@@ -17,17 +17,30 @@
*/
package org.apache.storm.kafka.spout;
-import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
-import org.junit.Test;
-
-import java.util.HashMap;
-
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.nullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
+import org.hamcrest.CoreMatchers;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
public class KafkaSpoutConfigTest {
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
@Test
public void testBasic() {
KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic").build();
@@ -35,17 +48,194 @@ public class KafkaSpoutConfigTest {
assertNull(conf.getConsumerGroupId());
assertTrue(conf.getTranslator() instanceof DefaultRecordTranslator);
HashMap<String, Object> expected = new HashMap<>();
- expected.put("bootstrap.servers", "localhost:1234");
- expected.put("enable.auto.commit", "false");
+ expected.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234");
+ expected.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+ expected.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ expected.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ expected.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
assertEquals(expected, conf.getKafkaProps());
+ assertEquals(KafkaSpoutConfig.DEFAULT_METRICS_TIME_BUCKET_SIZE_SECONDS, conf.getMetricsTimeBucketSizeInSecs());
}
@Test
- public void test_setEmitNullTuples_true_true() {
+ public void testSetEmitNullTuplesToTrue() {
final KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
.setEmitNullTuples(true)
.build();
assertTrue("Failed to set emit null tuples to true", conf.isEmitNullTuples());
}
+
+ @Test
+ public void testShouldNotChangeAutoOffsetResetPolicyWhenNotUsingAtLeastOnce() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE)
+ .build();
+
+ assertThat("When at-least-once is not specified, the spout should use the Kafka default auto offset reset policy",
+ conf.getKafkaProps().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), nullValue());
+ }
+
+ @Test
+ public void testWillRespectExplicitAutoOffsetResetPolicy() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none")
+ .build();
+
+ assertThat("Should allow users to pick a different auto offset reset policy than the one recommended for the at-least-once processing guarantee",
+ (String)conf.getKafkaProps().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), is("none"));
+ }
+
+ @Test
+ public void testCanConfigureWithExplicitTrueBooleanAutoCommitMode() {
+ /*
+ * Since adding setProcessingGuarantee to KafkaSpoutConfig we don't want users to set "enable.auto.commit" in the consumer config,
+ * because setting the processing guarantee will do it automatically. For backward compatibility we need to be able to handle the
+ * property being set anyway for a few releases, and try to set a processing guarantee that corresponds to the property.
+ */
+
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
+ .build();
+
+ assertThat("When setting enable auto commit to true explicitly the spout should use the 'none' processing guarantee",
+ conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE));
+ }
+
+ @Test
+ public void testCanConfigureWithExplicitFalseBooleanAutoCommitMode() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false)
+ .build();
+
+ assertThat("When setting enable auto commit to false explicitly the spout should use the 'at-least-once' processing guarantee",
+ conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE));
+ }
+
+ @Test
+ public void testCanConfigureWithExplicitTrueStringAutoCommitMode() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
+ .build();
+
+ assertThat("When setting enable auto commit to true explicitly the spout should use the 'none' processing guarantee",
+ conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE));
+ }
+
+ @Test
+ public void testCanConfigureWithExplicitFalseStringAutoCommitMode() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
+ .build();
+
+ assertThat("When setting enable auto commit explicitly to false the spout should use the 'at-least-once' processing guarantee",
+ conf.getProcessingGuarantee(), is(KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE));
+ }
+
+ @Test
+ public void testCanGetKeyDeserializerWhenUsingDefaultBuilder() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .build();
+
+ assertThat("When using the default builder methods, the key deserializer should default to StringDeserializer",
+ conf.getKeyDeserializer(), instanceOf(StringDeserializer.class));
+ }
+
+ @Test
+ public void testCanGetValueDeserializerWhenUsingDefaultBuilder() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .build();
+
+ assertThat("When using the default builder methods, the value deserializer should default to StringDeserializer",
+ conf.getValueDeserializer(), instanceOf(StringDeserializer.class));
+ }
+
+ @Test
+ public void testCanOverrideDeprecatedDeserializerClassWithKafkaProps() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setKey(StringDeserializer.class)
+ .setValue(StringDeserializer.class)
+ .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .build();
+
+ assertThat("The last set key deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
+ assertThat("The last set value deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
+ }
+
+ private static class SerializableStringDeserializer implements SerializableDeserializer {
+
+ private final StringDeserializer delegate = new StringDeserializer();
+
+ @Override
+ public void configure(Map configs, boolean isKey) {
+ delegate.configure(configs, isKey);
+ }
+
+ @Override
+ public Object deserialize(String topic, byte[] data) {
+ return delegate.deserialize(topic, data);
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+ }
+
+ @Test
+ public void testCanOverrideDeprecatedDeserializerInstanceWithKafkaProps() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setKey(new SerializableStringDeserializer())
+ .setValue(new SerializableStringDeserializer())
+ .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .build();
+
+ assertThat("The last set key deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
+ assertThat("The last set value deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
+ }
+
+ @Test
+ public void testCanOverrideKafkaPropsWithDeprecatedDeserializerSetter() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setKey(new SerializableStringDeserializer())
+ .setValue(new SerializableStringDeserializer())
+ .build();
+
+ assertThat("The last set key deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
+ assertThat("The last set value deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
+ }
+
+ @Test
+ public void testCanMixOldAndNewDeserializerSetter() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setKey(new SerializableStringDeserializer())
+ .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class)
+ .setValue(new SerializableStringDeserializer())
+ .build();
+
+ assertThat("The last set key deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(ByteArrayDeserializer.class));
+ assertThat("The last set value deserializer should be used, regardless of how it is set",
+ conf.getKafkaProps().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG), CoreMatchers.<Object>equalTo(SerializableStringDeserializer.class));
+ }
+
+ @Test
+ public void testMetricsTimeBucketSizeInSecs() {
+ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig.builder("localhost:1234", "topic")
+ .setMetricsTimeBucketSizeInSecs(100)
+ .build();
+
+ assertEquals(100, conf.getMetricsTimeBucketSizeInSecs());
+ }
}