You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/10/17 00:38:34 UTC
kafka git commit: MINOR: update exception message for KIP-120
Repository: kafka
Updated Branches:
refs/heads/trunk ef4914520 -> 7a1466392
MINOR: update exception message for KIP-120
Author: Matthias J. Sax <ma...@confluent.io>
Reviewers: Guozhang Wang <wa...@gmail.com>
Closes #4078 from mjsax/hotfix-streams
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7a146639
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7a146639
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7a146639
Branch: refs/heads/trunk
Commit: 7a14663929ba1111ba05b57e47cf96fc55891a36
Parents: ef49145
Author: Matthias J. Sax <ma...@confluent.io>
Authored: Mon Oct 16 17:38:31 2017 -0700
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Mon Oct 16 17:38:31 2017 -0700
----------------------------------------------------------------------
.../org/apache/kafka/streams/processor/internals/StreamThread.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/7a146639/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 234d254..0d4657d 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
@@ -862,7 +862,7 @@ public class StreamThread extends Thread implements ThreadDataProvider {
if (originalReset == null || (!originalReset.equals("earliest") && !originalReset.equals("latest"))) {
final 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, ...)";
+ "policy via StreamsBuilder#stream(..., Consumed.with(Topology.AutoOffsetReset)) or StreamsBuilder#table(..., Consumed.with(Topology.AutoOffsetReset))";
throw new StreamsException(String.format(errorMessage, partition.topic(), partition.partition()), e);
}