kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-4114: Allow different offset reset strategies
Date Thu, 12 Jan 2017 04:30:50 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6508e63c7 -> 23dff4b04


KAFKA-4114: Allow different offset reset strategies

mjsax

Here's my first pass at finer grained auto offset reset strategies.

I've left TODO comments about whether we want to consider adding this to `KGroupedTable.aggregate` and `KStreamImpl` when re-partitioning a source.

Author: bbejeck <bbejeck@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #2007 from bbejeck/KAFKA-4114_allow_different_offset_reset_strategies


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/23dff4b0
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/23dff4b0
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/23dff4b0

Branch: refs/heads/trunk
Commit: 23dff4b04a4c4cd338b5f39c9bb7d384857c482c
Parents: 6508e63
Author: Bill Bejeck <bbejeck@gmail.com>
Authored: Wed Jan 11 20:30:47 2017 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed Jan 11 20:30:47 2017 -0800

----------------------------------------------------------------------
 .../kafka/streams/kstream/KStreamBuilder.java   | 118 +++++++++-
 .../kstream/internals/KGroupedTableImpl.java    |   1 -
 .../streams/processor/TopologyBuilder.java      | 225 ++++++++++++++++++-
 .../processor/internals/StreamThread.java       |  54 ++++-
 ...eamsFineGrainedAutoResetIntegrationTest.java | 223 ++++++++++++++++++
 .../streams/processor/TopologyBuilderTest.java  |   2 +-
 6 files changed, 595 insertions(+), 28 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
index 33085c9..b1c1cfb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
@@ -55,7 +55,22 @@ public class KStreamBuilder extends TopologyBuilder {
      * @return a {@link KStream} for the specified topics
      */
     public <K, V> KStream<K, V> stream(String... topics) {
-        return stream(null, null, topics);
+        return stream(null, null, null, topics);
+    }
+
+
+    /**
+     * Create a {@link KStream} instance from the specified topics.
+     * The default deserializers specified in the config are used.
+     * <p>
+     * If multiple topics are specified there are nor ordering guaranteed for records from different topics.
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     * @param topics    the topic names; must contain at least one topic name
+     * @return a {@link KStream} for the specified topics
+     */
+    public <K, V> KStream<K, V> stream(AutoOffsetReset offsetReset, String... topics) {
+        return stream(offsetReset, null, null, topics);
     }
 
 
@@ -70,10 +85,26 @@ public class KStreamBuilder extends TopologyBuilder {
      * @return a {@link KStream} for topics matching the regex pattern.
      */
     public <K, V> KStream<K, V> stream(Pattern topicPattern) {
-        return stream(null, null, topicPattern);
+        return stream(null, null, null, topicPattern);
+    }
+
+    /**
+     * Create a {@link KStream} instance from the specified Pattern.
+     * The default deserializers specified in the config are used.
+     * <p>
+     * If multiple topics are matched by the specified pattern, the created stream will read data from all of them,
+     * and there is no ordering guarantee between records from different topics
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     * @param topicPattern    the Pattern to match for topic names
+     * @return a {@link KStream} for topics matching the regex pattern.
+     */
+    public <K, V> KStream<K, V> stream(AutoOffsetReset offsetReset, Pattern topicPattern) {
+        return stream(offsetReset, null, null, topicPattern);
     }
 
 
+
     /**
      * Create a {@link KStream} instance from the specified topics.
      * <p>
@@ -87,13 +118,33 @@ public class KStreamBuilder extends TopologyBuilder {
      * @return a {@link KStream} for the specified topics
      */
     public <K, V> KStream<K, V> stream(Serde<K> keySerde, Serde<V> valSerde, String... topics) {
+        return stream(null, keySerde, valSerde, topics);
+    }
+
+
+    /**
+     * Create a {@link KStream} instance from the specified topics.
+     * <p>
+     * If multiple topics are specified there are nor ordering guaranteed for records from different topics.
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     *
+     * @param keySerde  key serde used to read this source {@link KStream},
+     *                  if not specified the default serde defined in the configs will be used
+     * @param valSerde  value serde used to read this source {@link KStream},
+     *                  if not specified the default serde defined in the configs will be used
+     * @param topics    the topic names; must contain at least one topic name
+     * @return a {@link KStream} for the specified topics
+     */
+    public <K, V> KStream<K, V> stream(AutoOffsetReset offsetReset, Serde<K> keySerde, Serde<V> valSerde, String... topics) {
         String name = newName(KStreamImpl.SOURCE_NAME);
 
-        addSource(name, keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topics);
+        addSource(offsetReset, name,  keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topics);
 
         return new KStreamImpl<>(this, name, Collections.singleton(name), false);
     }
 
+
     /**
      * Create a {@link KStream} instance from the specified Pattern.
      * <p>
@@ -108,9 +159,27 @@ public class KStreamBuilder extends TopologyBuilder {
      * @return a {@link KStream} for the specified topics
      */
     public <K, V> KStream<K, V> stream(Serde<K> keySerde, Serde<V> valSerde, Pattern topicPattern) {
+        return stream(null, keySerde, valSerde, topicPattern);
+    }
+
+    /**
+     * Create a {@link KStream} instance from the specified Pattern.
+     * <p>
+     * If multiple topics are matched by the specified pattern, the created stream will read data from all of them,
+     * and there is no ordering guarantee between records from different topics.
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     * @param keySerde  key serde used to read this source {@link KStream},
+     *                  if not specified the default serde defined in the configs will be used
+     * @param valSerde  value serde used to read this source {@link KStream},
+     *                  if not specified the default serde defined in the configs will be used
+     * @param topicPattern    the Pattern to match for topic names
+     * @return a {@link KStream} for the specified topics
+     */
+    public <K, V> KStream<K, V> stream(AutoOffsetReset offsetReset, Serde<K> keySerde, Serde<V> valSerde, Pattern topicPattern) {
         String name = newName(KStreamImpl.SOURCE_NAME);
 
-        addSource(name, keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topicPattern);
+        addSource(offsetReset, name, keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topicPattern);
 
         return new KStreamImpl<>(this, name, Collections.singleton(name), false);
     }
@@ -122,14 +191,31 @@ public class KStreamBuilder extends TopologyBuilder {
      * The resulting {@link KTable} will be materialized in a local state store with the given store name.
      * However, no new changelog topic is created in this case since the underlying topic acts as one.
      *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     * @param topic     the topic name; cannot be null
+     * @param storeName the state store name used if this KTable is materialized, can be null if materialization not expected
+     * @return a {@link KTable} for the specified topics
+     */
+    public <K, V> KTable<K, V> table(AutoOffsetReset offsetReset, String topic, final String storeName) {
+        return table(offsetReset, null, null, topic, storeName);
+    }
+
+    /**
+     * Create a {@link KTable} instance for the specified topic.
+     * Record keys of the topic should never by null, otherwise an exception will be thrown at runtime.
+     * The default deserializers specified in the config are used.
+     * The resulting {@link KTable} will be materialized in a local state store with the given store name.
+     * However, no new changelog topic is created in this case since the underlying topic acts as one.
+     *
      * @param topic     the topic name; cannot be null
      * @param storeName the state store name used if this KTable is materialized, can be null if materialization not expected
      * @return a {@link KTable} for the specified topics
      */
     public <K, V> KTable<K, V> table(String topic, final String storeName) {
-        return table(null, null, topic, storeName);
+        return table(null, null, null, topic, storeName);
     }
 
+
     /**
      * Create a {@link KTable} instance for the specified topic.
      * Record keys of the topic should never by null, otherwise an exception will be thrown at runtime.
@@ -145,11 +231,30 @@ public class KStreamBuilder extends TopologyBuilder {
      * @return a {@link KTable} for the specified topics
      */
     public <K, V> KTable<K, V> table(Serde<K> keySerde, Serde<V> valSerde, String topic, final String storeName) {
+        return table(null, keySerde, valSerde, topic, storeName);
+    }
+
+    /**
+     * Create a {@link KTable} instance for the specified topic.
+     * Record keys of the topic should never by null, otherwise an exception will be thrown at runtime.
+     * The resulting {@link KTable} will be materialized in a local state store with the given store name.
+     * However, no new changelog topic is created in this case since the underlying topic acts as one.
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets available; acceptable values are earliest or latest
+     * @param keySerde   key serde used to send key-value pairs,
+     *                   if not specified the default key serde defined in the configuration will be used
+     * @param valSerde   value serde used to send key-value pairs,
+     *                   if not specified the default value serde defined in the configuration will be used
+     * @param topic      the topic name; cannot be null
+     * @param storeName  the state store name used if this KTable is materialized, can be null if materialization not expected
+     * @return a {@link KTable} for the specified topics
+     */
+    public <K, V> KTable<K, V> table(AutoOffsetReset offsetReset, Serde<K> keySerde, Serde<V> valSerde, String topic, final String storeName) {
         final String source = newName(KStreamImpl.SOURCE_NAME);
         final String name = newName(KTableImpl.SOURCE_NAME);
         final ProcessorSupplier<K, V> processorSupplier = new KTableSource<>(storeName);
 
-        addSource(source, keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topic);
+        addSource(offsetReset, source, keySerde == null ? null : keySerde.deserializer(), valSerde == null ? null : valSerde.deserializer(), topic);
         addProcessor(name, processorSupplier, source);
 
         final KTableImpl kTable = new KTableImpl<>(this, name, processorSupplier, Collections.singleton(source), storeName);
@@ -166,6 +271,7 @@ public class KStreamBuilder extends TopologyBuilder {
         return kTable;
     }
 
+
     /**
      * Create a new instance of {@link KStream} by merging the given streams.
      * <p>

http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
index 4edfa89..03fbbce 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java
@@ -68,7 +68,6 @@ public class KGroupedTableImpl<K, V> extends AbstractStream<K> implements KGroup
         return aggregate(initializer, adder, subtractor, keyValueStore(keySerde, aggValueSerde, storeName));
     }
 
-
     @Override
     public <T> KTable<K, T> aggregate(Initializer<T> initializer,
                                       Aggregator<? super K, ? super V, T> adder,

http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
index 211ed64..97cb4be 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
@@ -21,6 +21,8 @@ import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.errors.TopologyBuilderException;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.processor.internals.InternalTopicConfig;
 import org.apache.kafka.streams.processor.internals.ProcessorNode;
 import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
@@ -93,6 +95,16 @@ public class TopologyBuilder {
     // this is used in the extended KStreamBuilder.
     private final HashMap<String, String> sourceStoreToSourceTopic = new HashMap<>();
 
+    private final Set<String> earliestResetTopics = new HashSet<>();
+
+    private final Set<String> latestResetTopics = new HashSet<>();
+
+    private final Set<Pattern> earliestResetPatterns = new HashSet<>();
+
+    private final Set<Pattern> latestResetPatterns = new HashSet<>();
+
+    private static final Pattern EMPTY_ZERO_LENGTH_PATTERN = Pattern.compile("");
+
     private final QuickUnion<String> nodeGrouper = new QuickUnion<>();
 
     private SubscriptionUpdates subscriptionUpdates = new SubscriptionUpdates();
@@ -279,6 +291,13 @@ public class TopologyBuilder {
     }
 
     /**
+     * Enum used to define auto offset reset policy when creating {@link KStream} or {@link KTable}
+     */
+    public enum AutoOffsetReset {
+        EARLIEST , LATEST
+    }
+
+    /**
      * Create a new builder.
      */
     public TopologyBuilder() {}
@@ -311,7 +330,23 @@ public class TopologyBuilder {
      * @return this builder instance so methods can be chained together; never null
      */
     public synchronized final TopologyBuilder addSource(String name, String... topics) {
-        return addSource(name, (Deserializer) null, (Deserializer) null, topics);
+        return addSource(null, name, (Deserializer) null, (Deserializer) null, topics);
+    }
+
+    /**
+     * Add a new source that consumes the named topics and forward the records to child processor and/or sink nodes.
+     * The source will use the {@link org.apache.kafka.streams.StreamsConfig#KEY_SERDE_CLASS_CONFIG default key deserializer} and
+     * {@link org.apache.kafka.streams.StreamsConfig#VALUE_SERDE_CLASS_CONFIG default value deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}.
+     *
+     * @param offsetReset the auto offset reset policy to use for this source if no committed offsets found; acceptable values earliest or latest
+     * @param name the unique name of the source used to reference this node when
+     * {@link #addProcessor(String, ProcessorSupplier, String...) adding processor children}.
+     * @param topics the name of one or more Kafka topics that this source is to consume
+     * @return this builder instance so methods can be chained together; never null
+     */
+    public synchronized final TopologyBuilder addSource(AutoOffsetReset offsetReset, String name,  String... topics) {
+        return addSource(offsetReset, name, (Deserializer) null, (Deserializer) null, topics);
     }
 
 
@@ -328,9 +363,27 @@ public class TopologyBuilder {
      * @return this builder instance so methods can be chained together; never null
      */
     public synchronized final TopologyBuilder addSource(String name, Pattern topicPattern) {
-        return addSource(name, (Deserializer) null, (Deserializer) null, topicPattern);
+        return addSource(null, name, (Deserializer) null, (Deserializer) null, topicPattern);
+    }
+
+    /**
+     * Add a new source that consumes from topics matching the given pattern
+     * and forward the records to child processor and/or sink nodes.
+     * The source will use the {@link org.apache.kafka.streams.StreamsConfig#KEY_SERDE_CLASS_CONFIG default key deserializer} and
+     * {@link org.apache.kafka.streams.StreamsConfig#VALUE_SERDE_CLASS_CONFIG default value deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}.
+     *
+     * @param offsetReset the auto offset reset policy value for this source if no committed offsets found; acceptable values earliest or latest.
+     * @param name the unique name of the source used to reference this node when
+     * {@link #addProcessor(String, ProcessorSupplier, String...) adding processor children}.
+     * @param topicPattern regular expression pattern to match Kafka topics that this source is to consume
+     * @return this builder instance so methods can be chained together; never null
+     */
+    public synchronized final TopologyBuilder addSource(AutoOffsetReset offsetReset, String name,  Pattern topicPattern) {
+        return addSource(offsetReset, name, (Deserializer) null, (Deserializer) null, topicPattern);
     }
 
+
     /**
      * Add a new source that consumes the named topics and forwards the records to child processor and/or sink nodes.
      * The source will use the specified key and value deserializers.
@@ -348,6 +401,28 @@ public class TopologyBuilder {
      * @throws TopologyBuilderException if processor is already added or if topics have already been registered by another source
      */
     public synchronized final TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics) {
+        return addSource(null, name, keyDeserializer, valDeserializer, topics);
+
+    }
+
+    /**
+     * Add a new source that consumes the named topics and forwards the records to child processor and/or sink nodes.
+     * The source will use the specified key and value deserializers.
+     *
+     * @param offsetReset the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest.
+     * @param name the unique name of the source used to reference this node when
+     * {@link #addProcessor(String, ProcessorSupplier, String...) adding processor children}.
+     * @param keyDeserializer the {@link Deserializer key deserializer} used when consuming records; may be null if the source
+     * should use the {@link org.apache.kafka.streams.StreamsConfig#KEY_SERDE_CLASS_CONFIG default key deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}
+     * @param valDeserializer the {@link Deserializer value deserializer} used when consuming records; may be null if the source
+     * should use the {@link org.apache.kafka.streams.StreamsConfig#VALUE_SERDE_CLASS_CONFIG default value deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}
+     * @param topics the name of one or more Kafka topics that this source is to consume
+     * @return this builder instance so methods can be chained together; never null
+     * @throws TopologyBuilderException if processor is already added or if topics have already been registered by another source
+     */
+    public synchronized final TopologyBuilder addSource(AutoOffsetReset offsetReset, String name, Deserializer keyDeserializer, Deserializer valDeserializer, String... topics) {
         if (topics.length == 0) {
             throw new TopologyBuilderException("You must provide at least one topic");
         }
@@ -366,6 +441,9 @@ public class TopologyBuilder {
                 }
             }
 
+
+            maybeAddToResetList(earliestResetTopics, latestResetTopics, offsetReset, topic);
+
             sourceTopicNames.add(topic);
         }
 
@@ -397,6 +475,32 @@ public class TopologyBuilder {
      */
 
     public synchronized final TopologyBuilder addSource(String name, Deserializer keyDeserializer, Deserializer valDeserializer, Pattern topicPattern) {
+        return addSource(null, name,  keyDeserializer, valDeserializer, topicPattern);
+
+    }
+
+    /**
+     * Add a new source that consumes from topics matching the given pattern
+     * and forwards the records to child processor and/or sink nodes.
+     * The source will use the specified key and value deserializers. The provided
+     * de-/serializers will be used for all matched topics, so care should be taken to specify patterns for
+     * topics that share the same key-value data format.
+     *
+     * @param offsetReset  the auto offset reset policy to use for this stream if no committed offsets found; acceptable values are earliest or latest
+     * @param name the unique name of the source used to reference this node when
+     * {@link #addProcessor(String, ProcessorSupplier, String...) adding processor children}.
+     * @param keyDeserializer the {@link Deserializer key deserializer} used when consuming records; may be null if the source
+     * should use the {@link org.apache.kafka.streams.StreamsConfig#KEY_SERDE_CLASS_CONFIG default key deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}
+     * @param valDeserializer the {@link Deserializer value deserializer} used when consuming records; may be null if the source
+     * should use the {@link org.apache.kafka.streams.StreamsConfig#VALUE_SERDE_CLASS_CONFIG default value deserializer} specified in the
+     * {@link org.apache.kafka.streams.StreamsConfig stream configuration}
+     * @param topicPattern regular expression pattern to match Kafka topics that this source is to consume
+     * @return this builder instance so methods can be chained together; never null
+     * @throws TopologyBuilderException if processor is already added or if topics have already been registered by name
+     */
+
+    public synchronized final TopologyBuilder addSource(AutoOffsetReset offsetReset, String name,  Deserializer keyDeserializer, Deserializer valDeserializer, Pattern topicPattern) {
         Objects.requireNonNull(topicPattern, "topicPattern can't be null");
         Objects.requireNonNull(name, "name can't be null");
 
@@ -410,6 +514,8 @@ public class TopologyBuilder {
             }
         }
 
+        maybeAddToResetList(earliestResetPatterns, latestResetPatterns, offsetReset, topicPattern);
+
         nodeFactories.put(name, new SourceNodeFactory(name, null, topicPattern, keyDeserializer, valDeserializer));
         nodeToSourcePatterns.put(name, topicPattern);
         nodeGrouper.add(name);
@@ -722,6 +828,22 @@ public class TopologyBuilder {
                 Collections.unmodifiableSet(sourceTopics));
     }
 
+
+    private <T> void maybeAddToResetList(Collection<T> earliestResets, Collection<T> latestResets, AutoOffsetReset offsetReset, T item) {
+        if (offsetReset != null) {
+            switch (offsetReset) {
+                case EARLIEST:
+                    earliestResets.add(item);
+                    break;
+                case LATEST:
+                    latestResets.add(item);
+                    break;
+                default:
+                    throw new TopologyBuilderException(String.format("Unrecognized reset format %s", offsetReset));
+            }
+        }
+    }
+
     /**
      * Returns the map of node groups keyed by the topic group id.
      *
@@ -850,7 +972,7 @@ public class TopologyBuilder {
 
         return new ProcessorTopology(processorNodes, topicSourceMap, topicSinkMap, new ArrayList<>(stateStoreMap.values()), sourceStoreToSourceTopic, storeToProcessorNodeMap);
     }
-    
+
     /**
      * Returns the map of topic groups keyed by the group id.
      * A topic group is a group of topics in the same task.
@@ -954,6 +1076,84 @@ public class TopologyBuilder {
     }
 
     /**
+     * Get the Pattern to match all topics requiring to start reading from earliest available offset
+     * @return the Pattern for matching all topics reading from earliest offset, never null
+     */
+    public synchronized Pattern earliestResetTopicsPattern() {
+        Set<String> topics = maybeDecorateInternalSourceTopics(earliestResetTopics);
+
+        String[] sourceTopicNames = topics.toArray(new String[topics.size()]);
+        Pattern[] sourceTopicPatterns = earliestResetPatterns.toArray(new Pattern[earliestResetPatterns.size()]);
+
+        Pattern earliestPattern =  buildPatternForOffsetResetTopics(sourceTopicNames, sourceTopicPatterns);
+
+        ensureNoRegexOverlap(earliestPattern, latestResetPatterns, latestResetTopics);
+
+        return earliestPattern;
+    }
+
+    /**
+     * Get the Pattern to match all topics requiring to start reading from latest available offset
+     * @return the Pattern for matching all topics reading from latest offset, never null
+     */
+    public synchronized Pattern latestResetTopicsPattern() {
+        Set<String> topics = maybeDecorateInternalSourceTopics(latestResetTopics);
+
+        String[] sourceTopicNames = topics.toArray(new String[topics.size()]);
+        Pattern[] sourceTopicPatterns = latestResetPatterns.toArray(new Pattern[latestResetPatterns.size()]);
+
+        Pattern latestPattern = buildPatternForOffsetResetTopics(sourceTopicNames, sourceTopicPatterns);
+
+        ensureNoRegexOverlap(latestPattern, earliestResetPatterns, earliestResetTopics);
+
+        return  latestPattern;
+    }
+
+    private void ensureNoRegexOverlap(Pattern builtPattern, Set<Pattern> otherPatterns, Set<String> otherTopics) {
+
+        for (Pattern otherPattern : otherPatterns) {
+            if (builtPattern.pattern().contains(otherPattern.pattern())) {
+                throw new TopologyBuilderException(String.format("Found overlapping regex [%s] against [%s] for a KStream with auto offset resets", otherPattern.pattern(), builtPattern.pattern()));
+            }
+        }
+
+        for (String otherTopic : otherTopics) {
+            if (builtPattern.matcher(otherTopic).matches()) {
+                throw new TopologyBuilderException(String.format("Found overlapping regex [%s] matching topic [%s] for a KStream with auto offset resets", builtPattern.pattern(), otherTopic));
+            }
+        }
+
+    }
+
+
+    /**
+     * Builds a composite pattern out of topic names and Pattern object for matching topic names.  If the provided
+     * arrays are empty a Pattern.compile("") instance is returned.
+     *
+     * @param sourceTopics  the name of source topics to add to a composite pattern
+     * @param sourcePatterns Patterns for matching source topics to add to a composite pattern
+     * @return a Pattern that is composed of the literal source topic names and any Patterns for matching source topics
+     */
+    private static synchronized Pattern buildPatternForOffsetResetTopics(String[] sourceTopics, Pattern[] sourcePatterns) {
+        StringBuilder builder = new StringBuilder();
+
+        for (String topic : sourceTopics) {
+            builder.append(topic).append("|");
+        }
+
+        for (Pattern sourcePattern : sourcePatterns) {
+            builder.append(sourcePattern.pattern()).append("|");
+        }
+
+        if (builder.length() > 0) {
+            builder.setLength(builder.length() - 1);
+            return Pattern.compile(builder.toString());
+        }
+
+        return EMPTY_ZERO_LENGTH_PATTERN;
+    }
+
+    /**
      * @return a mapping from state store name to a Set of source Topics.
      */
     public Map<String, Set<String>> stateStoreNameToSourceTopics() {
@@ -1013,20 +1213,21 @@ public class TopologyBuilder {
 
     public synchronized Pattern sourceTopicPattern() {
         if (this.topicPattern == null && !nodeToSourcePatterns.isEmpty()) {
-            StringBuilder builder = new StringBuilder();
-            for (Pattern pattern : nodeToSourcePatterns.values()) {
-                builder.append(pattern.pattern()).append("|");
-            }
+
+            List<String> allNodeToSourceTopics = new ArrayList<>();
             if (!nodeToSourceTopics.isEmpty()) {
                 for (String[] topics : nodeToSourceTopics.values()) {
-                    for (String topic : topics) {
-                        builder.append(topic).append("|");
-                    }
+                    allNodeToSourceTopics.addAll(Arrays.asList(topics));
+
                 }
             }
+            int numPatterns = nodeToSourcePatterns.values().size();
+            int numTopics = allNodeToSourceTopics.size();
 
-            builder.setLength(builder.length() - 1);
-            this.topicPattern = Pattern.compile(builder.toString());
+            Pattern[] patterns = nodeToSourcePatterns.values().toArray(new Pattern[numPatterns]);
+            String[] allTopics = allNodeToSourceTopics.toArray(new String[numTopics]);
+
+            this.topicPattern = buildPatternForOffsetResetTopics(allTopics, patterns);
         }
         return this.topicPattern;
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 90194c6..5641849 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -19,9 +19,11 @@ package org.apache.kafka.streams.processor.internals;
 
 import org.apache.kafka.clients.consumer.CommitFailedException;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
@@ -199,6 +201,7 @@ public class StreamThread extends Thread {
     private final long commitTimeMs;
     private final StreamsMetricsThreadImpl streamsMetrics;
     final StateDirectory stateDirectory;
+    private String originalReset;
 
     private StreamPartitionAssignor partitionAssignor = null;
     private boolean cleanRun = false;
@@ -311,7 +314,16 @@ public class StreamThread extends Thread {
         log.info("{} Creating producer client", logPrefix);
         this.producer = clientSupplier.getProducer(config.getProducerConfigs(threadClientId));
         log.info("{} Creating consumer client", logPrefix);
-        this.consumer = clientSupplier.getConsumer(config.getConsumerConfigs(this, applicationId, threadClientId));
+
+        Map<String, Object> consumerConfigs = config.getConsumerConfigs(this, applicationId, threadClientId);
+
+        if (!builder.latestResetTopicsPattern().pattern().equals("") || !builder.earliestResetTopicsPattern().pattern().equals("")) {
+            originalReset = (String) consumerConfigs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG);
+            log.info("{} custom offset resets specified updating configs original auto offset reset {}", logPrefix, originalReset);
+            consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
+        }
+
+        this.consumer = clientSupplier.getConsumer(consumerConfigs);
         log.info("{} Creating restore consumer client", logPrefix);
         this.restoreConsumer = clientSupplier.getRestoreConsumer(config.getRestoreConsumerConfigs(threadClientId));
 
@@ -453,13 +465,8 @@ public class StreamThread extends Thread {
 
 
     /**
-<<<<<<< HEAD
      * Similar to shutdownTasksAndState, however does not close the task managers, in the hope that
      * soon the tasks will be assigned again
-=======
-     * Similar to shutdownTasksAndState, however does not close the task managers,
-     * in the hope that soon the tasks will be assigned again
->>>>>>> apache-kafka/trunk
      */
     private void suspendTasksAndState()  {
         log.debug("{} suspendTasksAndState: suspending all active tasks [{}] and standby tasks [{}]", logPrefix,
@@ -574,13 +581,44 @@ public class StreamThread extends Thread {
 
                 boolean longPoll = totalNumBuffered == 0;
 
-                ConsumerRecords<byte[], byte[]> records = consumer.poll(longPoll ? this.pollTimeMs : 0);
+                ConsumerRecords<byte[], byte[]> records = null;
+
+                try {
+                    records = consumer.poll(longPoll ? this.pollTimeMs : 0);
+                } catch (NoOffsetForPartitionException ex) {
+                    TopicPartition partition = ex.partition();
+                    if (builder.earliestResetTopicsPattern().matcher(partition.topic()).matches()) {
+                        log.info(String.format("stream-thread [%s] setting topic to consume from earliest offset %s", this.getName(), partition.topic()));
+                        consumer.seekToBeginning(ex.partitions());
+                    } else if (builder.latestResetTopicsPattern().matcher(partition.topic()).matches()) {
+                        consumer.seekToEnd(ex.partitions());
+                        log.info(String.format("stream-thread [%s] setting topic to consume from latest offset %s", this.getName(), partition.topic()));
+                    } else {
+
+                        if (originalReset == null || (!originalReset.equals("earliest") && !originalReset.equals("latest"))) {
+                            setState(State.PENDING_SHUTDOWN);
+                            String errorMessage = "No valid committed offset found for input topic %s (partition %s) and no valid reset policy configured." +
+                                    " You need to set configuration parameter \"auto.offset.reset\" or specify a topic specific reset " +
+                                    "policy via KStreamBuilder#stream(StreamsConfig.AutoOffsetReset offsetReset, ...) or KStreamBuilder#table(StreamsConfig.AutoOffsetReset offsetReset, ...)";
+                            throw new StreamsException(String.format(errorMessage, partition.topic(), partition.partition()), ex);
+                        }
+
+                        if (originalReset.equals("earliest")) {
+                            consumer.seekToBeginning(ex.partitions());
+                        } else if (originalReset.equals("latest")) {
+                            consumer.seekToEnd(ex.partitions());
+                        }
+                        log.info(String.format("stream-thread [%s] no custom setting defined for topic %s using original config %s for offset reset", this.getName(), partition.topic(), originalReset));
+                    }
+
+                }
 
                 if (rebalanceException != null)
                     throw new StreamsException(logPrefix + " Failed to rebalance", rebalanceException);
 
-                if (!records.isEmpty()) {
+                if (records != null && !records.isEmpty()) {
                     int numAddedRecords = 0;
+
                     for (TopicPartition partition : records.partitions()) {
                         StreamTask task = activeTasksByPartition.get(partition);
                         numAddedRecords += task.addRecords(partition, records.records(partition));

http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/test/java/org/apache/kafka/streams/integration/KStreamsFineGrainedAutoResetIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamsFineGrainedAutoResetIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamsFineGrainedAutoResetIntegrationTest.java
new file mode 100644
index 0000000..2171de1
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamsFineGrainedAutoResetIntegrationTest.java
@@ -0,0 +1,223 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.kafka.streams.integration;
+
+
+import kafka.utils.MockTime;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.TopologyBuilderException;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestCondition;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.regex.Pattern;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public class KStreamsFineGrainedAutoResetIntegrationTest {
+
+    private static final int NUM_BROKERS = 1;
+    private static final String DEFAULT_OUTPUT_TOPIC = "outputTopic";
+
+    @ClassRule
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
+    private final MockTime mockTime = CLUSTER.time;
+
+    private static final String TOPIC_1 = "topic-1";
+    private static final String TOPIC_2 = "topic-2";
+    private static final String TOPIC_A = "topic-A";
+    private static final String TOPIC_C = "topic-C";
+    private static final String TOPIC_Y = "topic-Y";
+    private static final String TOPIC_Z = "topic-Z";
+    private static final String NOOP = "noop";
+    private final Serde<String> stringSerde = Serdes.String();
+
+    private static final String STRING_SERDE_CLASSNAME = Serdes.String().getClass().getName();
+    private Properties streamsConfiguration;
+
+
+    @BeforeClass
+    public static void startKafkaCluster() throws Exception {
+        CLUSTER.createTopic(TOPIC_1);
+        CLUSTER.createTopic(TOPIC_2);
+        CLUSTER.createTopic(TOPIC_A);
+        CLUSTER.createTopic(TOPIC_C);
+        CLUSTER.createTopic(TOPIC_Y);
+        CLUSTER.createTopic(TOPIC_Z);
+        CLUSTER.createTopic(NOOP);
+        CLUSTER.createTopic(DEFAULT_OUTPUT_TOPIC);
+
+    }
+
+    @Before
+    public void setUp() throws Exception {
+
+        Properties props = new Properties();
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        streamsConfiguration = StreamsTestUtils.getStreamsConfig(
+                "testAutoOffsetId",
+                CLUSTER.bootstrapServers(),
+                STRING_SERDE_CLASSNAME,
+                STRING_SERDE_CLASSNAME,
+                props);
+
+        // Remove any state from previous test runs
+        IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
+    }
+
+    @Test
+    public void shouldOnlyReadRecordsWhereEarliestSpecified() throws  Exception {
+        final KStreamBuilder builder = new KStreamBuilder();
+
+        final KStream<String, String> pattern1Stream = builder.stream(KStreamBuilder.AutoOffsetReset.EARLIEST, Pattern.compile("topic-\\d"));
+        final KStream<String, String> pattern2Stream = builder.stream(KStreamBuilder.AutoOffsetReset.LATEST, Pattern.compile("topic-[A-D]"));
+        final KStream<String, String> namedTopicsStream = builder.stream(TOPIC_Y, TOPIC_Z);
+
+        pattern1Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+        pattern2Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+        namedTopicsStream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+
+        final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class);
+
+        final String topic1TestMessage = "topic-1 test";
+        final String topic2TestMessage = "topic-2 test";
+        final String topicATestMessage = "topic-A test";
+        final String topicCTestMessage = "topic-C test";
+        final String topicYTestMessage = "topic-Y test";
+        final String topicZTestMessage = "topic-Z test";
+
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_1, Collections.singletonList(topic1TestMessage), producerConfig, mockTime);
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_2, Collections.singletonList(topic2TestMessage), producerConfig, mockTime);
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_A, Collections.singletonList(topicATestMessage), producerConfig, mockTime);
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_C, Collections.singletonList(topicCTestMessage), producerConfig, mockTime);
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_Y, Collections.singletonList(topicYTestMessage), producerConfig, mockTime);
+        IntegrationTestUtils.produceValuesSynchronously(TOPIC_Z, Collections.singletonList(topicZTestMessage), producerConfig, mockTime);
+
+        final Properties consumerConfig = TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class);
+
+        final KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
+        streams.start();
+
+        final List<String> expectedReceivedValues = Arrays.asList(topic1TestMessage, topic2TestMessage, topicYTestMessage, topicZTestMessage);
+        final List<KeyValue<String, String>> receivedKeyValues = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, DEFAULT_OUTPUT_TOPIC, 4);
+        final List<String> actualValues = new ArrayList<>(4);
+
+        for (final KeyValue<String, String> receivedKeyValue : receivedKeyValues) {
+            actualValues.add(receivedKeyValue.value);
+        }
+
+        streams.close();
+        Collections.sort(actualValues);
+        Collections.sort(expectedReceivedValues);
+        assertThat(actualValues, equalTo(expectedReceivedValues));
+
+    }
+
+
+    @Test(expected = TopologyBuilderException.class)
+    public void shouldThrowExceptionOverlappingPattern() throws  Exception {
+        final KStreamBuilder builder = new KStreamBuilder();
+        //NOTE this would realistically get caught when building topology, the test is for completeness
+        final KStream<String, String> pattern1Stream = builder.stream(KStreamBuilder.AutoOffsetReset.EARLIEST, Pattern.compile("topic-[A-D]"));
+        final KStream<String, String> pattern2Stream = builder.stream(KStreamBuilder.AutoOffsetReset.LATEST, Pattern.compile("topic-[A-D]"));
+        final KStream<String, String> namedTopicsStream = builder.stream(TOPIC_Y, TOPIC_Z);
+
+        builder.earliestResetTopicsPattern();
+
+    }
+
+    @Test(expected = TopologyBuilderException.class)
+    public void shouldThrowExceptionOverlappingTopic() throws  Exception {
+        final KStreamBuilder builder = new KStreamBuilder();
+        //NOTE this would realistically get caught when building topology, the test is for completeness
+        final KStream<String, String> pattern1Stream = builder.stream(KStreamBuilder.AutoOffsetReset.EARLIEST, Pattern.compile("topic-[A-D]"));
+        final KStream<String, String> pattern2Stream = builder.stream(KStreamBuilder.AutoOffsetReset.LATEST, Pattern.compile("topic-\\d]"));
+        final KStream<String, String> namedTopicsStream = builder.stream(KStreamBuilder.AutoOffsetReset.LATEST, TOPIC_A, TOPIC_Z);
+
+        builder.latestResetTopicsPattern();
+
+    }
+
+
+    @Test
+    public void shouldThrowStreamsExceptionNoResetSpecified() throws Exception {
+        Properties props = new Properties();
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
+
+        Properties localConfig = StreamsTestUtils.getStreamsConfig(
+                "testAutoOffsetWithNone",
+                CLUSTER.bootstrapServers(),
+                STRING_SERDE_CLASSNAME,
+                STRING_SERDE_CLASSNAME,
+                props);
+
+        final KStreamBuilder builder = new KStreamBuilder();
+        final KStream<String, String> exceptionStream = builder.stream(NOOP);
+
+        exceptionStream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+
+        KafkaStreams streams = new KafkaStreams(builder, localConfig);
+
+        final TestingUncaughtExceptionHandler uncaughtExceptionHandler = new TestingUncaughtExceptionHandler();
+
+        final TestCondition correctExceptionThrownCondition = new TestCondition() {
+            @Override
+            public boolean conditionMet() {
+                return uncaughtExceptionHandler.correctExceptionThrown;
+            }
+        };
+
+        streams.setUncaughtExceptionHandler(uncaughtExceptionHandler);
+        streams.start();
+        TestUtils.waitForCondition(correctExceptionThrownCondition, "The expected NoOffsetForPartitionException was never thrown");
+        streams.close();
+    }
+
+
+    private static final class TestingUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+        boolean correctExceptionThrown = false;
+        @Override
+        public void uncaughtException(Thread t, Throwable e) {
+            assertThat(e.getClass().getSimpleName(), is("StreamsException"));
+            assertThat(e.getCause().getClass().getSimpleName(), is("NoOffsetForPartitionException"));
+            correctExceptionThrown = true;
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/23dff4b0/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
index c402c9b..b085a84 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java
@@ -184,7 +184,7 @@ public class TopologyBuilderTest {
     @Test
     public void testSubscribeTopicNameAndPattern() {
         final TopologyBuilder builder = new TopologyBuilder();
-        Pattern expectedPattern = Pattern.compile(".*-\\d|topic-foo|topic-bar");
+        Pattern expectedPattern = Pattern.compile("topic-foo|topic-bar|.*-\\d");
         builder.addSource("source-1", "topic-foo", "topic-bar");
         builder.addSource("source-2", Pattern.compile(".*-\\d"));
         assertEquals(expectedPattern.pattern(), builder.sourceTopicPattern().pattern());


Mime
View raw message