[jira] [Commented] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16813012#comment-16813012
]
Mohan Parthasarathy commented on KAFKA-5998:
I am not able to find the environemt field:
CentOS: 2.6.32-696
Kafka Version: 1.0.2
What else do you need ?
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Topology.txt, exc.txt,
> props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
[jira] [Commented] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16813009#comment-16813009
]
Guozhang Wang commented on KAFKA-5998:
--
This seems affecting lots of people's file systems.
Could everyone share your environment's OS / FS by editing this ticket's
{{Environment}} field (you can click "edit" above and search for that field),
while we can look deeper into it and try to find an ultimate fix? [~mparthas]
[~yaojingguo] [~j-white] [~dminkovsky] [~lbdai3190]
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Topology.txt, exc.txt,
> props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependenci
[jira] [Updated] (KAFKA-6399) Consider reducing "max.poll.interval.ms" default for Kafka Streams
[
https://issues.apache.org/jira/browse/KAFKA-6399?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-6399:
---
Description:
In Kafka {{0.10.2.1}} we change the default value of {{max.poll.intervall.ms}}
for Kafka Streams to {{Integer.MAX_VALUE}}. The reason was that long state
restore phases during rebalance could yield "rebalance storms" as consumers
drop out of a consumer group even if they are healthy as they didn't call
{{poll()}} during state restore phase.
In version {{0.11}} and {{1.0}} the state restore logic was improved a lot and
thus, now Kafka Streams does call {{poll()}} even during restore phase.
Therefore, we might consider setting a smaller timeout for
{{max.poll.intervall.ms}} to detect bad behaving Kafka Streams applications
(ie, targeting user code) that don't make progress any more during regular
operations.
The open question would be, what a good default might be. Maybe the actual
consumer default of 30 seconds might be sufficient. During one {{poll()}}
roundtrip, we would only call {{restoreConsumer.poll()}} once and restore a
single batch of records. This should take way less time than 30 seconds.
KIP-442:
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-442%3A+Return+to+default+max+poll+interval+in+Streams]
was:
In Kafka {{0.10.2.1}} we change the default value of {{max.poll.intervall.ms}}
for Kafka Streams to {{Integer.MAX_VALUE}}. The reason was that long state
restore phases during rebalance could yield "rebalance storms" as consumers
drop out of a consumer group even if they are healthy as they didn't call
{{poll()}} during state restore phase.
In version {{0.11}} and {{1.0}} the state restore logic was improved a lot and
thus, now Kafka Streams does call {{poll()}} even during restore phase.
Therefore, we might consider setting a smaller timeout for
{{max.poll.intervall.ms}} to detect bad behaving Kafka Streams applications
(ie, targeting user code) that don't make progress any more during regular
operations.
The open question would be, what a good default might be. Maybe the actual
consumer default of 30 seconds might be sufficient. During one {{poll()}}
roundtrip, we would only call {{restoreConsumer.poll()}} once and restore a
single batch of records. This should take way less time than 30 seconds.
> Consider reducing "max.poll.interval.ms" default for Kafka Streams
> --
>
> Key: KAFKA-6399
> URL: https://issues.apache.org/jira/browse/KAFKA-6399
> Project: Kafka
> Issue Type: Improvement
> Components: streams
>Affects Versions: 1.0.0
>Reporter: Matthias J. Sax
>Assignee: John Roesler
>Priority: Minor
> Labels: kip
> Fix For: 2.3.0
>
>
> In Kafka {{0.10.2.1}} we change the default value of
> {{max.poll.intervall.ms}} for Kafka Streams to {{Integer.MAX_VALUE}}. The
> reason was that long state restore phases during rebalance could yield
> "rebalance storms" as consumers drop out of a consumer group even if they are
> healthy as they didn't call {{poll()}} during state restore phase.
> In version {{0.11}} and {{1.0}} the state restore logic was improved a lot
> and thus, now Kafka Streams does call {{poll()}} even during restore phase.
> Therefore, we might consider setting a smaller timeout for
> {{max.poll.intervall.ms}} to detect bad behaving Kafka Streams applications
> (ie, targeting user code) that don't make progress any more during regular
> operations.
> The open question would be, what a good default might be. Maybe the actual
> consumer default of 30 seconds might be sufficient. During one {{poll()}}
> roundtrip, we would only call {{restoreConsumer.poll()}} once and restore a
> single batch of records. This should take way less time than 30 seconds.
> KIP-442:
> [https://cwiki.apache.org/confluence/display/KAFKA/KIP-442%3A+Return+to+default+max+poll+interval+in+Streams]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-6399) Consider reducing "max.poll.interval.ms" default for Kafka Streams
[
https://issues.apache.org/jira/browse/KAFKA-6399?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-6399:
---
Labels: kip (was: )
> Consider reducing "max.poll.interval.ms" default for Kafka Streams
> --
>
> Key: KAFKA-6399
> URL: https://issues.apache.org/jira/browse/KAFKA-6399
> Project: Kafka
> Issue Type: Improvement
> Components: streams
>Affects Versions: 1.0.0
>Reporter: Matthias J. Sax
>Assignee: John Roesler
>Priority: Minor
> Labels: kip
> Fix For: 2.3.0
>
>
> In Kafka {{0.10.2.1}} we change the default value of
> {{max.poll.intervall.ms}} for Kafka Streams to {{Integer.MAX_VALUE}}. The
> reason was that long state restore phases during rebalance could yield
> "rebalance storms" as consumers drop out of a consumer group even if they are
> healthy as they didn't call {{poll()}} during state restore phase.
> In version {{0.11}} and {{1.0}} the state restore logic was improved a lot
> and thus, now Kafka Streams does call {{poll()}} even during restore phase.
> Therefore, we might consider setting a smaller timeout for
> {{max.poll.intervall.ms}} to detect bad behaving Kafka Streams applications
> (ie, targeting user code) that don't make progress any more during regular
> operations.
> The open question would be, what a good default might be. Maybe the actual
> consumer default of 30 seconds might be sufficient. During one {{poll()}}
> roundtrip, we would only call {{restoreConsumer.poll()}} once and restore a
> single batch of records. This should take way less time than 30 seconds.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Guozhang Wang updated KAFKA-5998:
-
Priority: Critical (was: Major)
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Priority: Critical
> Attachments: 5998.v1.txt, 5998.v2.txt, Topology.txt, exc.txt,
> props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.j
[jira] [Updated] (KAFKA-8200) TopologyTestDriver should offer an iterable signature of readOutput
[
https://issues.apache.org/jira/browse/KAFKA-8200?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-8200:
---
Labels: needs-kip (was: )
> TopologyTestDriver should offer an iterable signature of readOutput
> ---
>
> Key: KAFKA-8200
> URL: https://issues.apache.org/jira/browse/KAFKA-8200
> Project: Kafka
> Issue Type: Improvement
> Components: streams
>Reporter: Michael Drogalis
>Priority: Minor
> Labels: needs-kip
>
> When using the TopologyTestDriver, one examines the output on a topic with
> the readOutput method. This method returns one record at a time, until no
> more records can be found, at which point in returns null.
> Many times, the usage pattern around readOutput will involve writing a loop
> to extract a number of records from the topic, building up a list of records,
> until it returns null.
> It would be helpful to offer an iterable signature of readOutput, which
> returns either an iterator or list over the records that are currently
> available in the topic. This would effectively remove the loop that a user
> needs to write for him/herself each time.
> Such a signature might look like:
> {code:java}
> public Iterable> readOutput(java.lang.String
> topic);
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8200) TopologyTestDriver should offer an iterable signature of readOutput
[
https://issues.apache.org/jira/browse/KAFKA-8200?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-8200:
---
Component/s: streams
> TopologyTestDriver should offer an iterable signature of readOutput
> ---
>
> Key: KAFKA-8200
> URL: https://issues.apache.org/jira/browse/KAFKA-8200
> Project: Kafka
> Issue Type: Improvement
> Components: streams
>Reporter: Michael Drogalis
>Priority: Minor
>
> When using the TopologyTestDriver, one examines the output on a topic with
> the readOutput method. This method returns one record at a time, until no
> more records can be found, at which point in returns null.
> Many times, the usage pattern around readOutput will involve writing a loop
> to extract a number of records from the topic, building up a list of records,
> until it returns null.
> It would be helpful to offer an iterable signature of readOutput, which
> returns either an iterator or list over the records that are currently
> available in the topic. This would effectively remove the loop that a user
> needs to write for him/herself each time.
> Such a signature might look like:
> {code:java}
> public Iterable> readOutput(java.lang.String
> topic);
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-4600) Consumer proceeds on when ConsumerRebalanceListener fails
[ https://issues.apache.org/jira/browse/KAFKA-4600?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812848#comment-16812848 ] Guozhang Wang commented on KAFKA-4600: -- [~braedon] I think your argument makes sense. But we still need to define, that when such exception gets thrown to user's face from `consumer.poll` call, and users decide to continue and retry instead of shutting down the consumer, what should happen -- back to my example above, as in option 2), how to treat partitions that failed in revocation / assignment. So how about this: let me piggy-back this proposal along with KIP-429 since it will change the behavior of revocation and assignment orderings anyways. After I'm done adding a section regarding the error handling I'll ping you on this ticket and we can continue our discussion to fix it in the next release. > Consumer proceeds on when ConsumerRebalanceListener fails > - > > Key: KAFKA-4600 > URL: https://issues.apache.org/jira/browse/KAFKA-4600 > Project: Kafka > Issue Type: Bug > Components: consumer >Affects Versions: 0.10.1.1 >Reporter: Braedon Vickers >Priority: Major > > One of the use cases for a ConsumerRebalanceListener is to load state > necessary for processing a partition when it is assigned. However, when > ConsumerRebalanceListener.onPartitionsAssigned() fails for some reason (i.e. > the state isn't loaded), the error is logged and the consumer proceeds on as > if nothing happened, happily consuming messages from the new partition. When > the state is relied upon for correct processing, this can be very bad, e.g. > data loss can occur. > It would be better if the error was propagated up so it could be dealt with > normally. At the very least the assignment should fail so the consumer > doesn't see any messages from the new partitions, and the rebalance can be > reattempted. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8108) Flaky Test kafka.api.ClientIdQuotaTest.testThrottledProducerConsumer
[
https://issues.apache.org/jira/browse/KAFKA-8108?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812847#comment-16812847
]
Guozhang Wang commented on KAFKA-8108:
--
Happened again:
https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/3618/testReport/junit/kafka.api/UserQuotaTest/testThrottledProducerConsumer/
{code}
java.lang.AssertionError: Client with id=QuotasTestProducer-1 should have been
throttled
at org.junit.Assert.fail(Assert.java:89)
at org.junit.Assert.assertTrue(Assert.java:42)
at
kafka.api.QuotaTestClients.verifyThrottleTimeMetric(BaseQuotaTest.scala:230)
at
kafka.api.QuotaTestClients.verifyProduceThrottle(BaseQuotaTest.scala:216)
at
kafka.api.BaseQuotaTest.testThrottledProducerConsumer(BaseQuotaTest.scala:83){code}
> Flaky Test kafka.api.ClientIdQuotaTest.testThrottledProducerConsumer
>
>
> Key: KAFKA-8108
> URL: https://issues.apache.org/jira/browse/KAFKA-8108
> Project: Kafka
> Issue Type: Improvement
>Reporter: Guozhang Wang
>Priority: Major
>
> {code}
> java.lang.AssertionError: Client with id=QuotasTestProducer-!@#$%^&*() should
> have been throttled
> at org.junit.Assert.fail(Assert.java:89)
> at org.junit.Assert.assertTrue(Assert.java:42)
> at
> kafka.api.QuotaTestClients.verifyThrottleTimeMetric(BaseQuotaTest.scala:229)
> at
> kafka.api.QuotaTestClients.verifyProduceThrottle(BaseQuotaTest.scala:215)
> at
> kafka.api.BaseQuotaTest.testThrottledProducerConsumer(BaseQuotaTest.scala:82)
> {code}
> https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/3230/testReport/junit/kafka.api/ClientIdQuotaTest/testThrottledProducerConsumer/
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-8200) TopologyTestDriver should offer an iterable signature of readOutput
Michael Drogalis created KAFKA-8200:
---
Summary: TopologyTestDriver should offer an iterable signature of
readOutput
Key: KAFKA-8200
URL: https://issues.apache.org/jira/browse/KAFKA-8200
Project: Kafka
Issue Type: Improvement
Reporter: Michael Drogalis
When using the TopologyTestDriver, one examines the output on a topic with the
readOutput method. This method returns one record at a time, until no more
records can be found, at which point in returns null.
Many times, the usage pattern around readOutput will involve writing a loop to
extract a number of records from the topic, building up a list of records,
until it returns null.
It would be helpful to offer an iterable signature of readOutput, which returns
either an iterator or list over the records that are currently available in the
topic. This would effectively remove the loop that a user needs to write for
him/herself each time.
Such a signature might look like:
{code:java}
public Iterable> readOutput(java.lang.String
topic);
{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-7656) ReplicaManager fetch fails on leader due to long/integer overflow
[
https://issues.apache.org/jira/browse/KAFKA-7656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812825#comment-16812825
]
Jose Armando Garcia Sancio edited comment on KAFKA-7656 at 4/8/19 9:51 PM:
---
Also, [~pachilo], can you turn on request logging as documented here:
https://issues.apache.org/jira/browse/KAFKA-7656?focusedCommentId=16757766&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-16757766
was (Author: jagsancio):
Also, [~pachilo], you turn on request logging as documented here:
https://issues.apache.org/jira/browse/KAFKA-7656?focusedCommentId=16757766&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-16757766
> ReplicaManager fetch fails on leader due to long/integer overflow
> -
>
> Key: KAFKA-7656
> URL: https://issues.apache.org/jira/browse/KAFKA-7656
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.0.1
> Environment: Linux 3.10.0-693.el7.x86_64 #1 SMP Thu Jul 6 19:56:57
> EDT 2017 x86_64 x86_64 x86_64 GNU/Linux
>Reporter: Patrick Haas
>Assignee: Jose Armando Garcia Sancio
>Priority: Major
>
> (Note: From 2.0.1-cp1 from confluent distribution)
> {{[2018-11-19 21:13:13,687] ERROR [ReplicaManager broker=103] Error
> processing fetch operation on partition __consumer_offsets-20, offset 0
> (kafka.server.ReplicaManager)}}
> {{java.lang.IllegalArgumentException: Invalid max size -2147483648 for log
> read from segment FileRecords(file=
> /prod/kafka/data/kafka-logs/__consumer_offsets-20/.log,
> start=0, end=2147483647)}}
> {{ at kafka.log.LogSegment.read(LogSegment.scala:274)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1159)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1114)}}
> {{ at kafka.log.Log.maybeHandleIOException(Log.scala:1842)}}
> {{ at kafka.log.Log.read(Log.scala:1114)}}
> {{ at
> kafka.server.ReplicaManager.kafka$server$ReplicaManager$$read$1(ReplicaManager.scala:912)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:974)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:973)}}
> {{ at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)}}
> {{ at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)}}
> {{ at kafka.server.ReplicaManager.readFromLocalLog(ReplicaManager.scala:973)}}
> {{ at kafka.server.ReplicaManager.readFromLog$1(ReplicaManager.scala:802)}}
> {{ at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:815)}}
> {{ at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:685)}}
> {{ at kafka.server.KafkaApis.handle(KafkaApis.scala:114)}}
> {{ at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)}}
> {{ at java.lang.Thread.run(Thread.java:748)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7656) ReplicaManager fetch fails on leader due to long/integer overflow
[
https://issues.apache.org/jira/browse/KAFKA-7656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812825#comment-16812825
]
Jose Armando Garcia Sancio commented on KAFKA-7656:
---
Also, [~pachilo], you turn on request logging as documented here:
https://issues.apache.org/jira/browse/KAFKA-7656?focusedCommentId=16757766&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-16757766
> ReplicaManager fetch fails on leader due to long/integer overflow
> -
>
> Key: KAFKA-7656
> URL: https://issues.apache.org/jira/browse/KAFKA-7656
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.0.1
> Environment: Linux 3.10.0-693.el7.x86_64 #1 SMP Thu Jul 6 19:56:57
> EDT 2017 x86_64 x86_64 x86_64 GNU/Linux
>Reporter: Patrick Haas
>Assignee: Jose Armando Garcia Sancio
>Priority: Major
>
> (Note: From 2.0.1-cp1 from confluent distribution)
> {{[2018-11-19 21:13:13,687] ERROR [ReplicaManager broker=103] Error
> processing fetch operation on partition __consumer_offsets-20, offset 0
> (kafka.server.ReplicaManager)}}
> {{java.lang.IllegalArgumentException: Invalid max size -2147483648 for log
> read from segment FileRecords(file=
> /prod/kafka/data/kafka-logs/__consumer_offsets-20/.log,
> start=0, end=2147483647)}}
> {{ at kafka.log.LogSegment.read(LogSegment.scala:274)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1159)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1114)}}
> {{ at kafka.log.Log.maybeHandleIOException(Log.scala:1842)}}
> {{ at kafka.log.Log.read(Log.scala:1114)}}
> {{ at
> kafka.server.ReplicaManager.kafka$server$ReplicaManager$$read$1(ReplicaManager.scala:912)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:974)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:973)}}
> {{ at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)}}
> {{ at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)}}
> {{ at kafka.server.ReplicaManager.readFromLocalLog(ReplicaManager.scala:973)}}
> {{ at kafka.server.ReplicaManager.readFromLog$1(ReplicaManager.scala:802)}}
> {{ at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:815)}}
> {{ at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:685)}}
> {{ at kafka.server.KafkaApis.handle(KafkaApis.scala:114)}}
> {{ at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)}}
> {{ at java.lang.Thread.run(Thread.java:748)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7656) ReplicaManager fetch fails on leader due to long/integer overflow
[
https://issues.apache.org/jira/browse/KAFKA-7656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812823#comment-16812823
]
Jose Armando Garcia Sancio commented on KAFKA-7656:
---
[~pachilo], thanks for the update. Can you please share the relevant logs from
the leader and follower that is causing this issue?
> ReplicaManager fetch fails on leader due to long/integer overflow
> -
>
> Key: KAFKA-7656
> URL: https://issues.apache.org/jira/browse/KAFKA-7656
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.0.1
> Environment: Linux 3.10.0-693.el7.x86_64 #1 SMP Thu Jul 6 19:56:57
> EDT 2017 x86_64 x86_64 x86_64 GNU/Linux
>Reporter: Patrick Haas
>Assignee: Jose Armando Garcia Sancio
>Priority: Major
>
> (Note: From 2.0.1-cp1 from confluent distribution)
> {{[2018-11-19 21:13:13,687] ERROR [ReplicaManager broker=103] Error
> processing fetch operation on partition __consumer_offsets-20, offset 0
> (kafka.server.ReplicaManager)}}
> {{java.lang.IllegalArgumentException: Invalid max size -2147483648 for log
> read from segment FileRecords(file=
> /prod/kafka/data/kafka-logs/__consumer_offsets-20/.log,
> start=0, end=2147483647)}}
> {{ at kafka.log.LogSegment.read(LogSegment.scala:274)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1159)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1114)}}
> {{ at kafka.log.Log.maybeHandleIOException(Log.scala:1842)}}
> {{ at kafka.log.Log.read(Log.scala:1114)}}
> {{ at
> kafka.server.ReplicaManager.kafka$server$ReplicaManager$$read$1(ReplicaManager.scala:912)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:974)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:973)}}
> {{ at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)}}
> {{ at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)}}
> {{ at kafka.server.ReplicaManager.readFromLocalLog(ReplicaManager.scala:973)}}
> {{ at kafka.server.ReplicaManager.readFromLog$1(ReplicaManager.scala:802)}}
> {{ at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:815)}}
> {{ at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:685)}}
> {{ at kafka.server.KafkaApis.handle(KafkaApis.scala:114)}}
> {{ at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)}}
> {{ at java.lang.Thread.run(Thread.java:748)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (KAFKA-8181) Streams docs on serialization include Avro header, but no content
[ https://issues.apache.org/jira/browse/KAFKA-8181?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax reassigned KAFKA-8181: -- Assignee: Victoria Bialas > Streams docs on serialization include Avro header, but no content > - > > Key: KAFKA-8181 > URL: https://issues.apache.org/jira/browse/KAFKA-8181 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.2.0 >Reporter: Michael Drogalis >Assignee: Victoria Bialas >Priority: Minor > Labels: documentation > Fix For: 2.3.0 > > > On [the documentation for data types and > serialization|https://kafka.apache.org/10/documentation/streams/developer-guide/datatypes.html], > Avro is listed in the table of contents as something supported out of the > box. The link is dead, though, because there is no content. We should either > remove the header or supply the content. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7656) ReplicaManager fetch fails on leader due to long/integer overflow
[
https://issues.apache.org/jira/browse/KAFKA-7656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812809#comment-16812809
]
Jonathan Santilli commented on KAFKA-7656:
--
I facing the same issue, but just after updating from version 2.0 to version 2.2
I can confirm that is the replica fetcher.
I have stopped all producer and consumers and stopped the brokers one by one
until the message stopped. After bringing back all brokers, the error is still
showing up.
> ReplicaManager fetch fails on leader due to long/integer overflow
> -
>
> Key: KAFKA-7656
> URL: https://issues.apache.org/jira/browse/KAFKA-7656
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.0.1
> Environment: Linux 3.10.0-693.el7.x86_64 #1 SMP Thu Jul 6 19:56:57
> EDT 2017 x86_64 x86_64 x86_64 GNU/Linux
>Reporter: Patrick Haas
>Assignee: Jose Armando Garcia Sancio
>Priority: Major
>
> (Note: From 2.0.1-cp1 from confluent distribution)
> {{[2018-11-19 21:13:13,687] ERROR [ReplicaManager broker=103] Error
> processing fetch operation on partition __consumer_offsets-20, offset 0
> (kafka.server.ReplicaManager)}}
> {{java.lang.IllegalArgumentException: Invalid max size -2147483648 for log
> read from segment FileRecords(file=
> /prod/kafka/data/kafka-logs/__consumer_offsets-20/.log,
> start=0, end=2147483647)}}
> {{ at kafka.log.LogSegment.read(LogSegment.scala:274)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1159)}}
> {{ at kafka.log.Log$$anonfun$read$2.apply(Log.scala:1114)}}
> {{ at kafka.log.Log.maybeHandleIOException(Log.scala:1842)}}
> {{ at kafka.log.Log.read(Log.scala:1114)}}
> {{ at
> kafka.server.ReplicaManager.kafka$server$ReplicaManager$$read$1(ReplicaManager.scala:912)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:974)}}
> {{ at
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:973)}}
> {{ at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)}}
> {{ at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)}}
> {{ at kafka.server.ReplicaManager.readFromLocalLog(ReplicaManager.scala:973)}}
> {{ at kafka.server.ReplicaManager.readFromLog$1(ReplicaManager.scala:802)}}
> {{ at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:815)}}
> {{ at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:685)}}
> {{ at kafka.server.KafkaApis.handle(KafkaApis.scala:114)}}
> {{ at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)}}
> {{ at java.lang.Thread.run(Thread.java:748)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-8013) Avoid buffer underflow when reading a Struct from a partially correct buffer
[
https://issues.apache.org/jira/browse/KAFKA-8013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812804#comment-16812804
]
ASF GitHub Bot commented on KAFKA-8013:
---
hachikuji commented on pull request #6340: KAFKA-8013: Avoid underflow when
reading a Struct from a partially correct buffer
URL: https://github.com/apache/kafka/pull/6340
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Avoid buffer underflow when reading a Struct from a partially correct buffer
>
>
> Key: KAFKA-8013
> URL: https://issues.apache.org/jira/browse/KAFKA-8013
> Project: Kafka
> Issue Type: Bug
>Reporter: Konstantine Karantasis
>Assignee: Konstantine Karantasis
>Priority: Major
> Fix For: 2.3.0
>
>
> Protocol compatibility can be facilitated if a {{Struct}}, that has been
> defined as an extension of a previous {{Struct}} by adding fields at the end
> of the older version, can read an older version by ignoring the absence of
> the missing new fields. Of course this has to be allowed by the definition of
> these fields (they have to be {{nullable}}).
> For example, this should work:
> {code:java}
> Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
> Schema newSchema = new Schema(new Field("field1", Type.NULLABLE_STRING), new
> Field("field2" , Type.NULLABLE_STRING));
> String value = "foo bar baz";
> Struct oldFormat = new Struct(oldSchema).set("field1", value);
> ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
> oldFormat.writeTo(buffer);
> buffer.flip();
> Struct newFormat = newSchema.read(buffer);
> assertEquals(value, newFormat.get("field1"));
> assertEquals(null, newFormat.get("field2"));
> {code}
> Currently it does not.
> A fix to the above is considered safe, because depending on buffer underflow
> to detect missing data at the end of a {{Struct}} is not an appropriate
> check.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-8013) Avoid buffer underflow when reading a Struct from a partially correct buffer
[
https://issues.apache.org/jira/browse/KAFKA-8013?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jason Gustafson resolved KAFKA-8013.
Resolution: Fixed
> Avoid buffer underflow when reading a Struct from a partially correct buffer
>
>
> Key: KAFKA-8013
> URL: https://issues.apache.org/jira/browse/KAFKA-8013
> Project: Kafka
> Issue Type: Bug
>Reporter: Konstantine Karantasis
>Assignee: Konstantine Karantasis
>Priority: Major
> Fix For: 2.3.0
>
>
> Protocol compatibility can be facilitated if a {{Struct}}, that has been
> defined as an extension of a previous {{Struct}} by adding fields at the end
> of the older version, can read an older version by ignoring the absence of
> the missing new fields. Of course this has to be allowed by the definition of
> these fields (they have to be {{nullable}}).
> For example, this should work:
> {code:java}
> Schema oldSchema = new Schema(new Field("field1", Type.NULLABLE_STRING));
> Schema newSchema = new Schema(new Field("field1", Type.NULLABLE_STRING), new
> Field("field2" , Type.NULLABLE_STRING));
> String value = "foo bar baz";
> Struct oldFormat = new Struct(oldSchema).set("field1", value);
> ByteBuffer buffer = ByteBuffer.allocate(oldSchema.sizeOf(oldFormat));
> oldFormat.writeTo(buffer);
> buffer.flip();
> Struct newFormat = newSchema.read(buffer);
> assertEquals(value, newFormat.get("field1"));
> assertEquals(null, newFormat.get("field2"));
> {code}
> Currently it does not.
> A fix to the above is considered safe, because depending on buffer underflow
> to detect missing data at the end of a {{Struct}} is not an appropriate
> check.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-6399) Consider reducing "max.poll.interval.ms" default for Kafka Streams
[
https://issues.apache.org/jira/browse/KAFKA-6399?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-6399:
---
Fix Version/s: 2.3.0
> Consider reducing "max.poll.interval.ms" default for Kafka Streams
> --
>
> Key: KAFKA-6399
> URL: https://issues.apache.org/jira/browse/KAFKA-6399
> Project: Kafka
> Issue Type: Improvement
> Components: streams
>Affects Versions: 1.0.0
>Reporter: Matthias J. Sax
>Assignee: John Roesler
>Priority: Minor
> Fix For: 2.3.0
>
>
> In Kafka {{0.10.2.1}} we change the default value of
> {{max.poll.intervall.ms}} for Kafka Streams to {{Integer.MAX_VALUE}}. The
> reason was that long state restore phases during rebalance could yield
> "rebalance storms" as consumers drop out of a consumer group even if they are
> healthy as they didn't call {{poll()}} during state restore phase.
> In version {{0.11}} and {{1.0}} the state restore logic was improved a lot
> and thus, now Kafka Streams does call {{poll()}} even during restore phase.
> Therefore, we might consider setting a smaller timeout for
> {{max.poll.intervall.ms}} to detect bad behaving Kafka Streams applications
> (ie, targeting user code) that don't make progress any more during regular
> operations.
> The open question would be, what a good default might be. Maybe the actual
> consumer default of 30 seconds might be sufficient. During one {{poll()}}
> roundtrip, we would only call {{restoreConsumer.poll()}} once and restore a
> single batch of records. This should take way less time than 30 seconds.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8174) Can't call arbitrary SimpleBenchmarks tests from streams_simple_benchmark_test.py
[ https://issues.apache.org/jira/browse/KAFKA-8174?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-8174: --- Component/s: system tests streams > Can't call arbitrary SimpleBenchmarks tests from > streams_simple_benchmark_test.py > - > > Key: KAFKA-8174 > URL: https://issues.apache.org/jira/browse/KAFKA-8174 > Project: Kafka > Issue Type: Bug > Components: streams, system tests >Reporter: Sophie Blee-Goldman >Priority: Minor > > When using the script streams_simple_benchmark_test.py you should be able to > specify a test name and run that particular method in SimpleBenchmarks. This > works for most existing benchmarks, however you can't use this to run the > "yahoo" benchmark and you can't add new tests to SimpleBenchmarks and start > them successfully. > > If you try to run yahoo/new test it fails with the error "Not enough > parameters are provided; expecting propFileName, testName, numRecords, > keySkew, valueSize" in main(); the missing argument turns out to be testName. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-8174) Can't call arbitrary SimpleBenchmarks tests from streams_simple_benchmark_test.py
[ https://issues.apache.org/jira/browse/KAFKA-8174?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-8174: --- Issue Type: Improvement (was: Bug) > Can't call arbitrary SimpleBenchmarks tests from > streams_simple_benchmark_test.py > - > > Key: KAFKA-8174 > URL: https://issues.apache.org/jira/browse/KAFKA-8174 > Project: Kafka > Issue Type: Improvement > Components: streams, system tests >Reporter: Sophie Blee-Goldman >Priority: Minor > > When using the script streams_simple_benchmark_test.py you should be able to > specify a test name and run that particular method in SimpleBenchmarks. This > works for most existing benchmarks, however you can't use this to run the > "yahoo" benchmark and you can't add new tests to SimpleBenchmarks and start > them successfully. > > If you try to run yahoo/new test it fails with the error "Not enough > parameters are provided; expecting propFileName, testName, numRecords, > keySkew, valueSize" in main(); the missing argument turns out to be testName. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8194) MessagesInPerSec incorrect value when Stream produce messages
[
https://issues.apache.org/jira/browse/KAFKA-8194?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812790#comment-16812790
]
Matthias J. Sax commented on KAFKA-8194:
Seems the Jira title is miss leading? Can we update it? It's not a Kafka
Streams issue.
> MessagesInPerSec incorrect value when Stream produce messages
> -
>
> Key: KAFKA-8194
> URL: https://issues.apache.org/jira/browse/KAFKA-8194
> Project: Kafka
> Issue Type: Bug
> Components: metrics
>Affects Versions: 1.1.0, 2.2.0
>Reporter: Odyldzhon Toshbekov
>Priority: Trivial
> Attachments: Screen Shot 2019-04-05 at 17.51.03.png, Screen Shot
> 2019-04-05 at 17.52.22.png
>
>
> Looks like metric
> {code:java}
> kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec{code}
> has incorrect value when messages come via Kafka Stream API.
> I noticed that offset for every message from Kafka Stream can be increased by
> 1,2,... However if messages come to Broker from Kafka producer it's always
> incremented by 1.
> Unfortunately the metric mentioned above calculated based on offset changes
> and as result we cannot use streams because metric will be always incorrect.
> For Kafka 2.2.0
> !Screen Shot 2019-04-05 at 17.51.03.png|width=100%!
>
> [https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/server/ReplicaManager.scala]
> And this is the method used to get "numAppendedMessages"
> !Screen Shot 2019-04-05 at 17.52.22.png|width=100%!
> https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/log/Log.scala
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7895) Ktable supress operator emitting more than one record for the same key per window
[
https://issues.apache.org/jira/browse/KAFKA-7895?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812787#comment-16812787
]
John Roesler commented on KAFKA-7895:
-
No problem. Thanks for _your_ help!
The PR is being reviewed now. Once it's merged, I'll request it to be
cherry-picked to all versions that have Suppress, and then request bugfix
releases for them.
As an update, I've tracked down the problem where I'm still seeing duplicates
when I crash the application, even with EOS enabled. The problem is that
Streams flushes the state stores in no particular order. If there is a cached
state store upstream of the suppression, and Streams flushes the suppression
buffer first, and then the cached state store second, the cached data will get
flushed and processed by Suppress, and there may be processing results that
Suppress emits, but the changelog for the buffer won't be updated again. The
solution I have in mind is to flush the stores in topological order. In the
mean time, disabling caching should eliminate this particular source of
duplicates, if someone is looking for a workaround.
> Ktable supress operator emitting more than one record for the same key per
> window
> -
>
> Key: KAFKA-7895
> URL: https://issues.apache.org/jira/browse/KAFKA-7895
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.1.0, 2.1.1
>Reporter: prasanthi
>Assignee: John Roesler
>Priority: Major
> Fix For: 2.2.0, 2.1.2
>
>
> Hi, We are using kstreams to get the aggregated counts per vendor(key) within
> a specified window.
> Here's how we configured the suppress operator to emit one final record per
> key/window.
> {code:java}
> KTable, Long> windowedCount = groupedStream
> .windowedBy(TimeWindows.of(Duration.ofMinutes(1)).grace(ofMillis(5L)))
> .count(Materialized.with(Serdes.Integer(),Serdes.Long()))
> .suppress(Suppressed.untilWindowCloses(unbounded()));
> {code}
> But we are getting more than one record for the same key/window as shown
> below.
> {code:java}
> [KTABLE-TOSTREAM-10]: [131@154906704/154906710], 1039
> [KTABLE-TOSTREAM-10]: [131@154906704/154906710], 1162
> [KTABLE-TOSTREAM-10]: [9@154906704/154906710], 6584
> [KTABLE-TOSTREAM-10]: [88@154906704/154906710], 107
> [KTABLE-TOSTREAM-10]: [108@154906704/154906710], 315
> [KTABLE-TOSTREAM-10]: [119@154906704/154906710], 119
> [KTABLE-TOSTREAM-10]: [154@154906704/154906710], 746
> [KTABLE-TOSTREAM-10]: [154@154906704/154906710], 809{code}
> Could you please take a look?
> Thanks
>
>
> Added by John:
> Acceptance Criteria:
> * add suppress to system tests, such that it's exercised with crash/shutdown
> recovery, rebalance, etc.
> ** [https://github.com/apache/kafka/pull/6278]
> * make sure that there's some system test coverage with caching disabled.
> ** Follow-on ticket: https://issues.apache.org/jira/browse/KAFKA-7943
> * test with tighter time bounds with windows of say 30 seconds and use
> system time without adding any extra time for verification
> ** Follow-on ticket: https://issues.apache.org/jira/browse/KAFKA-7944
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8193) Flaky Test MetricsIntegrationTest#testStreamMetricOfWindowStore
[ https://issues.apache.org/jira/browse/KAFKA-8193?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-8193: --- Component/s: (was: admin) streams > Flaky Test MetricsIntegrationTest#testStreamMetricOfWindowStore > --- > > Key: KAFKA-8193 > URL: https://issues.apache.org/jira/browse/KAFKA-8193 > Project: Kafka > Issue Type: Bug > Components: streams, unit tests >Affects Versions: 2.3.0 >Reporter: Konstantine Karantasis >Assignee: Guozhang Wang >Priority: Critical > Labels: flaky-test > Fix For: 2.3.0 > > > [https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/3576/console] > *14:14:48* org.apache.kafka.streams.integration.MetricsIntegrationTest > > testStreamMetricOfWindowStore STARTED > *14:14:59* > org.apache.kafka.streams.integration.MetricsIntegrationTest.testStreamMetricOfWindowStore > failed, log available in > /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.12/streams/build/reports/testOutput/org.apache.kafka.streams.integration.MetricsIntegrationTest.testStreamMetricOfWindowStore.test.stdout > *14:14:59* > *14:14:59* org.apache.kafka.streams.integration.MetricsIntegrationTest > > testStreamMetricOfWindowStore FAILED > *14:14:59* java.lang.AssertionError: Condition not met within timeout 1. > testStoreMetricWindow -> Size of metrics of type:'put-latency-avg' must be > equal to:2 but it's equal to 0 expected:<2> but was:<0> > *14:14:59* at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:361) > *14:14:59* at > org.apache.kafka.streams.integration.MetricsIntegrationTest.testStreamMetricOfWindowStore(MetricsIntegrationTest.java:260) > *14:15:01* -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-8181) Streams docs on serialization include Avro header, but no content
[ https://issues.apache.org/jira/browse/KAFKA-8181?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax resolved KAFKA-8181. Resolution: Fixed Fix Version/s: 2.3.0 > Streams docs on serialization include Avro header, but no content > - > > Key: KAFKA-8181 > URL: https://issues.apache.org/jira/browse/KAFKA-8181 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.2.0 >Reporter: Michael Drogalis >Priority: Minor > Labels: documentation > Fix For: 2.3.0 > > > On [the documentation for data types and > serialization|https://kafka.apache.org/10/documentation/streams/developer-guide/datatypes.html], > Avro is listed in the table of contents as something supported out of the > box. The link is dead, though, because there is no content. We should either > remove the header or supply the content. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-8198) KStreams testing docs use non-existent method "pipe"
[
https://issues.apache.org/jira/browse/KAFKA-8198?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-8198:
---
Component/s: streams
documentation
> KStreams testing docs use non-existent method "pipe"
>
>
> Key: KAFKA-8198
> URL: https://issues.apache.org/jira/browse/KAFKA-8198
> Project: Kafka
> Issue Type: Bug
> Components: documentation, streams
>Affects Versions: 1.1.1, 2.0.1, 2.2.0, 2.1.1
>Reporter: Michael Drogalis
>Priority: Minor
> Labels: documentation, newbie
>
> In [the testing docs for
> KStreams|https://kafka.apache.org/20/documentation/streams/developer-guide/testing.html],
> we use the following code snippet:
> {code:java}
> ConsumerRecordFactory factory = new
> ConsumerRecordFactory<>("input-topic", new StringSerializer(), new
> IntegerSerializer());
> testDriver.pipe(factory.create("key", 42L));
> {code}
> As of Apache Kafka 2.2.0, this method no longer exists. We should correct the
> docs to use the pipeInput method.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8169) Wrong topic on streams quick start documentation
[ https://issues.apache.org/jira/browse/KAFKA-8169?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-8169: --- Component/s: streams > Wrong topic on streams quick start documentation > > > Key: KAFKA-8169 > URL: https://issues.apache.org/jira/browse/KAFKA-8169 > Project: Kafka > Issue Type: Improvement > Components: documentation, streams >Affects Versions: 2.1.1 >Reporter: Tcheutchoua Steve >Assignee: Tcheutchoua Steve >Priority: Minor > Labels: documentation > Fix For: 2.3.0 > > Original Estimate: 1h > Remaining Estimate: 1h > > [Kafka Streams Quick > Start|[https://kafka.apache.org/21/documentation/streams/quickstart]] > Though out the tutorial, the name of the input topic that was created is > `streams-plaintext-input`. However, this was mistaken at some point in the > tutorial and changed to `streams-wordcount-input`. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-8169) Wrong topic on streams quick start documentation
[ https://issues.apache.org/jira/browse/KAFKA-8169?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax resolved KAFKA-8169. Resolution: Fixed Fix Version/s: 2.3.0 [~tcheutchoua], I added you to the list of contributors and assigned the ticket to you. Thanks for the fix! > Wrong topic on streams quick start documentation > > > Key: KAFKA-8169 > URL: https://issues.apache.org/jira/browse/KAFKA-8169 > Project: Kafka > Issue Type: Improvement > Components: documentation >Affects Versions: 2.1.1 >Reporter: Tcheutchoua Steve >Assignee: Tcheutchoua Steve >Priority: Minor > Labels: documentation > Fix For: 2.3.0 > > Original Estimate: 1h > Remaining Estimate: 1h > > [Kafka Streams Quick > Start|[https://kafka.apache.org/21/documentation/streams/quickstart]] > Though out the tutorial, the name of the input topic that was created is > `streams-plaintext-input`. However, this was mistaken at some point in the > tutorial and changed to `streams-wordcount-input`. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-8169) Wrong topic on streams quick start documentation
[ https://issues.apache.org/jira/browse/KAFKA-8169?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax reassigned KAFKA-8169: -- Assignee: Tcheutchoua Steve > Wrong topic on streams quick start documentation > > > Key: KAFKA-8169 > URL: https://issues.apache.org/jira/browse/KAFKA-8169 > Project: Kafka > Issue Type: Improvement > Components: documentation >Affects Versions: 2.1.1 >Reporter: Tcheutchoua Steve >Assignee: Tcheutchoua Steve >Priority: Minor > Labels: documentation > Original Estimate: 1h > Remaining Estimate: 1h > > [Kafka Streams Quick > Start|[https://kafka.apache.org/21/documentation/streams/quickstart]] > Though out the tutorial, the name of the input topic that was created is > `streams-plaintext-input`. However, this was mistaken at some point in the > tutorial and changed to `streams-wordcount-input`. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-5998) /.checkpoint.tmp Not found exception
[
https://issues.apache.org/jira/browse/KAFKA-5998?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812743#comment-16812743
]
Mohan Parthasarathy commented on KAFKA-5998:
That did not seem to help much. We still see this error:
2019-04-08 19:38:50,074 WARN ProcessorStateManager: task [0_3] Failed to write
offset checkpoint file to /tmp/kafka-state/test-app/0_3/.checkpoint:
java.io.FileNotFoundException: /tmp/kafka-state/test-app/0_3/.checkpoint.tmp
(No such file or directory)
If i look under /tmp/kafka-state/test-app, most of the time you see only 1_X
directories. Sometimes I do see 0_X directories and then it disappears. Also,
not sure whether it is related to this or not, when new messages come, the old
messages are getting reprocessed. Not sure because of the check point problem.
> /.checkpoint.tmp Not found exception
>
>
> Key: KAFKA-5998
> URL: https://issues.apache.org/jira/browse/KAFKA-5998
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 0.11.0.0, 0.11.0.1, 2.1.1
>Reporter: Yogesh BG
>Priority: Major
> Attachments: 5998.v1.txt, 5998.v2.txt, Topology.txt, exc.txt,
> props.txt, streams.txt
>
>
> I have one kafka broker and one kafka stream running... I am running its
> since two days under load of around 2500 msgs per second.. On third day am
> getting below exception for some of the partitions, I have 16 partitions only
> 0_0 and 0_1 gives this error
> {{09:43:25.955 [ks_0_inst-StreamThread-6] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_1/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:221)
> ~[na:1.7.0_111]
> at java.io.FileOutputStream.(FileOutputStream.java:171)
> ~[na:1.7.0_111]
> at
> org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:73)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:324)
> ~[rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:267)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:201)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:260)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:254)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:322)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:415)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:314)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:700)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:683)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:523)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:480)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:457)
> [rtp-kafkastreams-1.0-SNAPSHOT-jar-with-dependencies.jar:na]
> 09:43:25.974 [ks_0_inst-StreamThread-15] WARN
> o.a.k.s.p.i.ProcessorStateManager - Failed to write checkpoint file to
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint:
> java.io.FileNotFoundException:
> /data/kstreams/rtp-kafkastreams/0_0/.checkpoint.tmp (No such file or
> directory)
> at java.io.FileOutputStream.open(Native Method) ~[na:1.7.0_111]
> at java.io.FileO
[jira] [Commented] (KAFKA-7841) KIP-419 Safely notify Kafka Connect SourceTask is stopped
[ https://issues.apache.org/jira/browse/KAFKA-7841?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812711#comment-16812711 ] ASF GitHub Bot commented on KAFKA-7841: --- AndrewJSchofield commented on pull request #6551: KAFKA-7841: Implement KIP-419 adding SourceTask.stopped method URL: https://github.com/apache/kafka/pull/6551 Add a new SourceTask.stopped method called as the last method in the lifecycle of a SourceTask. Called just as the resources are cleaned up in the Kafka Connect runtime. Testing by adding checks that the new method is called as expected in the existing Kafka Connect runtime tests. The contribution is my original work and I license the work to the project under the project's open source license. ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > KIP-419 Safely notify Kafka Connect SourceTask is stopped > - > > Key: KAFKA-7841 > URL: https://issues.apache.org/jira/browse/KAFKA-7841 > Project: Kafka > Issue Type: New Feature > Components: KafkaConnect >Affects Versions: 2.2.0 >Reporter: Andrew Schofield >Priority: Minor > Attachments: The.txt > > > Implements KIP 419. > https://cwiki.apache.org/confluence/display/KAFKA/KIP-419%3A+Safely+notify+Kafka+Connect+SourceTask+is+stopped -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-5784) Add a sensor for dropped records in window stores
[
https://issues.apache.org/jira/browse/KAFKA-5784?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Guozhang Wang resolved KAFKA-5784.
--
Resolution: Fixed
Fix Version/s: 2.1.0
> Add a sensor for dropped records in window stores
> -
>
> Key: KAFKA-5784
> URL: https://issues.apache.org/jira/browse/KAFKA-5784
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: siva santhalingam
>Priority: Major
> Labels: newbie
> Fix For: 2.1.0
>
>
> Today when a {{put(record)}} call on a windowed store does not find the
> corresponding segment, i.e. its corresponding's window has expired, we simply
> returns a {{null}} and hence silently drops it.
> We should consider 1) add log4j entries when it happens, 2) add metrics (we
> can discuss whether it should be a processor-node level, or store level
> sensor) for such cases.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-5784) Add a sensor for dropped records in window stores
[
https://issues.apache.org/jira/browse/KAFKA-5784?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812664#comment-16812664
]
Guozhang Wang commented on KAFKA-5784:
--
I think this is already added as part of [~vvcephei]'s PR on the suppression
operator.
> Add a sensor for dropped records in window stores
> -
>
> Key: KAFKA-5784
> URL: https://issues.apache.org/jira/browse/KAFKA-5784
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Reporter: Guozhang Wang
>Assignee: siva santhalingam
>Priority: Major
> Labels: newbie
>
> Today when a {{put(record)}} call on a windowed store does not find the
> corresponding segment, i.e. its corresponding's window has expired, we simply
> returns a {{null}} and hence silently drops it.
> We should consider 1) add log4j entries when it happens, 2) add metrics (we
> can discuss whether it should be a processor-node level, or store level
> sensor) for such cases.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-8194) MessagesInPerSec incorrect value when Stream produce messages
[
https://issues.apache.org/jira/browse/KAFKA-8194?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812629#comment-16812629
]
Guozhang Wang commented on KAFKA-8194:
--
[~odyldz...@gmail.com] I've added you to the confluence wiki space. Thanks!
> MessagesInPerSec incorrect value when Stream produce messages
> -
>
> Key: KAFKA-8194
> URL: https://issues.apache.org/jira/browse/KAFKA-8194
> Project: Kafka
> Issue Type: Bug
> Components: metrics
>Affects Versions: 1.1.0, 2.2.0
>Reporter: Odyldzhon Toshbekov
>Priority: Trivial
> Attachments: Screen Shot 2019-04-05 at 17.51.03.png, Screen Shot
> 2019-04-05 at 17.52.22.png
>
>
> Looks like metric
> {code:java}
> kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec{code}
> has incorrect value when messages come via Kafka Stream API.
> I noticed that offset for every message from Kafka Stream can be increased by
> 1,2,... However if messages come to Broker from Kafka producer it's always
> incremented by 1.
> Unfortunately the metric mentioned above calculated based on offset changes
> and as result we cannot use streams because metric will be always incorrect.
> For Kafka 2.2.0
> !Screen Shot 2019-04-05 at 17.51.03.png|width=100%!
>
> [https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/server/ReplicaManager.scala]
> And this is the method used to get "numAppendedMessages"
> !Screen Shot 2019-04-05 at 17.52.22.png|width=100%!
> https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/log/Log.scala
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-6474) Rewrite test to use new public TopologyTestDriver
[ https://issues.apache.org/jira/browse/KAFKA-6474?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812597#comment-16812597 ] Guozhang Wang commented on KAFKA-6474: -- Hi [~h314to] could you provide an update on the remaining of PRs you are planning on to remove the deprecated KStreamTestDriver? > Rewrite test to use new public TopologyTestDriver > - > > Key: KAFKA-6474 > URL: https://issues.apache.org/jira/browse/KAFKA-6474 > Project: Kafka > Issue Type: Improvement > Components: streams, unit tests >Affects Versions: 1.1.0 >Reporter: Matthias J. Sax >Assignee: Filipe Agapito >Priority: Major > Labels: newbie > > With KIP-247 we added public TopologyTestDriver. We should rewrite out own > test to use this new test driver and remove the two classes > ProcessorTopoogyTestDriver and KStreamTestDriver. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-8157) Missing "key.serializer" exception when setting "segment index bytes"
[
https://issues.apache.org/jira/browse/KAFKA-8157?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Guozhang Wang resolved KAFKA-8157.
--
Resolution: Fixed
Assignee: Guozhang Wang
Fix Version/s: 2.2.1
2.3.0
With the PR merged and cherry-picked to older branches as well, I'm resolving
this ticket now.
> Missing "key.serializer" exception when setting "segment index bytes"
> -
>
> Key: KAFKA-8157
> URL: https://issues.apache.org/jira/browse/KAFKA-8157
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.2.0
> Environment: ubuntu 18.10, localhost and Aiven too
>Reporter: Cristian D
>Assignee: Guozhang Wang
>Priority: Major
> Labels: beginner, newbie
> Fix For: 2.3.0, 2.2.1
>
>
> As a `kafka-streams` user,
> When I set the "segment index bytes" property
> Then I would like to have internal topics with the specified allocated disk
> space
>
> At the moment, when setting the "topic.segment.index.bytes" property, the
> application is exiting with following exception:
> {code:java}
> Exception in thread "main" org.apache.kafka.common.config.ConfigException:
> Missing required configuration "key.serializer" which has no default value.
> {code}
> Tested with `kafka-streams` v2.0.0 and v2.2.0.
>
> Stack trace:
> {code:java}
> Exception in thread "main" org.apache.kafka.common.config.ConfigException:
> Missing required configuration "key.serializer" which has no default value.
> at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:474)
> at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:464)
> at
> org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:62)
> at
> org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:75)
> at
> org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:392)
> at
> org.apache.kafka.streams.StreamsConfig.getMainConsumerConfigs(StreamsConfig.java:1014)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.create(StreamThread.java:666)
> at org.apache.kafka.streams.KafkaStreams.(KafkaStreams.java:718)
> at org.apache.kafka.streams.KafkaStreams.(KafkaStreams.java:634)
> at org.apache.kafka.streams.KafkaStreams.(KafkaStreams.java:544)
> at app.Main.main(Main.java:36)
> {code}
> A demo application simulating the exception:
> https://github.com/razorcd/java-snippets-and-demo-projects/tree/master/kafkastreamsdemo
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7983) supporting replication.throttled.replicas in dynamic broker configuration
[ https://issues.apache.org/jira/browse/KAFKA-7983?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812556#comment-16812556 ] Jun Rao commented on KAFKA-7983: [~rahul.mnit], just added you to the contributor list and assigned the ticket to you. Thanks. > supporting replication.throttled.replicas in dynamic broker configuration > - > > Key: KAFKA-7983 > URL: https://issues.apache.org/jira/browse/KAFKA-7983 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Jun Rao >Assignee: Rahul Agarwal >Priority: Major > > In > [KIP-226|https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration#KIP-226-DynamicBrokerConfiguration-DefaultTopicconfigs], > we added the support to change broker defaults dynamically. However, it > didn't support changing leader.replication.throttled.replicas and > follower.replication.throttled.replicas. These 2 configs were introduced in > [KIP-73|https://cwiki.apache.org/confluence/display/KAFKA/KIP-73+Replication+Quotas] > and controls the set of topic partitions on which replication throttling > will be engaged. One useful case is to be able to set a default value for > both configs to * to allow throttling to be engaged for all topic partitions. > Currently, the static default value for both configs are ignored for > replication throttling, it would be useful to fix that as well. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-7983) supporting replication.throttled.replicas in dynamic broker configuration
[ https://issues.apache.org/jira/browse/KAFKA-7983?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jun Rao reassigned KAFKA-7983: -- Assignee: Rahul Agarwal > supporting replication.throttled.replicas in dynamic broker configuration > - > > Key: KAFKA-7983 > URL: https://issues.apache.org/jira/browse/KAFKA-7983 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Jun Rao >Assignee: Rahul Agarwal >Priority: Major > > In > [KIP-226|https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration#KIP-226-DynamicBrokerConfiguration-DefaultTopicconfigs], > we added the support to change broker defaults dynamically. However, it > didn't support changing leader.replication.throttled.replicas and > follower.replication.throttled.replicas. These 2 configs were introduced in > [KIP-73|https://cwiki.apache.org/confluence/display/KAFKA/KIP-73+Replication+Quotas] > and controls the set of topic partitions on which replication throttling > will be engaged. One useful case is to be able to set a default value for > both configs to * to allow throttling to be engaged for all topic partitions. > Currently, the static default value for both configs are ignored for > replication throttling, it would be useful to fix that as well. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-8199) ClassCastException when trying to groupBy after suppress
[
https://issues.apache.org/jira/browse/KAFKA-8199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-8199:
---
Description:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder();
builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
was:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder();
builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
> ClassCastException when trying to groupBy after suppress
>
>
> Key: KAFKA-8199
> URL: https://issues.apache.org/jira/browse/KAFKA-8199
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.1.0
>Reporter: Bill Bejeck
>Priority: Major
> Fix For: 2.3.0
>
>
> A topology with a groupBy after a suppress operation results in a
> ClassCastException
> The following sample topology
> {noformat}
> Properties properties = new Properties();
> properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
> properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
> StreamsBuilder builder = new StreamsBuilder();
> builder.stream("topic")
> .groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
> .suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
> BufferConfig.unbounded()))
> .groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
> builder.build(properties);
> {noformat}
> results in this exception:
> {noformat}
> java.lang.ClassCastException:
> org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
> cannot be cast to
> org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8199) ClassCastException when trying to groupBy after suppress
[
https://issues.apache.org/jira/browse/KAFKA-8199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-8199:
---
Description:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder();
builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
was:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
> ClassCastException when trying to groupBy after suppress
>
>
> Key: KAFKA-8199
> URL: https://issues.apache.org/jira/browse/KAFKA-8199
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.1.0
>Reporter: Bill Bejeck
>Priority: Major
> Fix For: 2.3.0
>
>
> A topology with a groupBy after a suppress operation results in a
> ClassCastException
> The following sample topology
> {noformat}
> Properties properties = new Properties();
> properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
> properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
> StreamsBuilder builder = new StreamsBuilder();
> builder.stream("topic")
> .groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
> .suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
> BufferConfig.unbounded()))
> .groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
> builder.build(properties);
> {noformat}
> results in this exception:
> {noformat}
> java.lang.ClassCastException:
> org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
> cannot be cast to
> org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8199) ClassCastException when trying to groupBy after suppress
[
https://issues.apache.org/jira/browse/KAFKA-8199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-8199:
---
Description:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
was:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic").groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded())) .groupBy((k, v) ->
KeyValue.pair(k,v)).count().toStream(); builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
> ClassCastException when trying to groupBy after suppress
>
>
> Key: KAFKA-8199
> URL: https://issues.apache.org/jira/browse/KAFKA-8199
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.1.0
>Reporter: Bill Bejeck
>Priority: Major
> Fix For: 2.3.0
>
>
> A topology with a groupBy after a suppress operation results in a
> ClassCastException
> The following sample topology
> {noformat}
> Properties properties = new Properties();
> properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
> properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
> StreamsBuilder builder = new StreamsBuilder(); builder. String>stream("topic")
> .groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
> .suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
> BufferConfig.unbounded()))
> .groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
> builder.build(properties);
> {noformat}
> results in this exception:
> {noformat}
> java.lang.ClassCastException:
> org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
> cannot be cast to
> org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-8199) ClassCastException when trying to groupBy after suppress
[
https://issues.apache.org/jira/browse/KAFKA-8199?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bill Bejeck updated KAFKA-8199:
---
Description:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
was:
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic")
.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded()))
.groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
> ClassCastException when trying to groupBy after suppress
>
>
> Key: KAFKA-8199
> URL: https://issues.apache.org/jira/browse/KAFKA-8199
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.1.0
>Reporter: Bill Bejeck
>Priority: Major
> Fix For: 2.3.0
>
>
> A topology with a groupBy after a suppress operation results in a
> ClassCastException
> The following sample topology
> {noformat}
> Properties properties = new Properties();
> properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
> properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
> StreamsBuilder builder = new StreamsBuilder(); builder. String>stream("topic")
> .groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
> .suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
> BufferConfig.unbounded()))
> .groupBy((k, v) -> KeyValue.pair(k,v)).count().toStream();
> builder.build(properties);
> {noformat}
> results in this exception:
> {noformat}
> java.lang.ClassCastException:
> org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
> cannot be cast to
> org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-8199) ClassCastException when trying to groupBy after suppress
Bill Bejeck created KAFKA-8199:
--
Summary: ClassCastException when trying to groupBy after suppress
Key: KAFKA-8199
URL: https://issues.apache.org/jira/browse/KAFKA-8199
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 2.1.0
Reporter: Bill Bejeck
Fix For: 2.3.0
A topology with a groupBy after a suppress operation results in a
ClassCastException
The following sample topology
{noformat}
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "appid");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost");
StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic").groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))).count()
.suppress(Suppressed.untilTimeLimit(Duration.ofHours(1),
BufferConfig.unbounded())) .groupBy((k, v) ->
KeyValue.pair(k,v)).count().toStream(); builder.build(properties);
{noformat}
results in this exception:
{noformat}
java.lang.ClassCastException:
org.apache.kafka.streams.kstream.internals.KTableImpl$$Lambda$4/2084435065
cannot be cast to
org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier{noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7965) Flaky Test ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
[
https://issues.apache.org/jira/browse/KAFKA-7965?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812335#comment-16812335
]
huxihx commented on KAFKA-7965:
---
Seems [PR 6328|[https://github.com/apache/kafka/pull/6238]] changed the code a
lot. Still got the AssertionError right now?
> Flaky Test
> ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
>
>
> Key: KAFKA-7965
> URL: https://issues.apache.org/jira/browse/KAFKA-7965
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, unit tests
>Affects Versions: 2.2.0, 2.3.0
>Reporter: Matthias J. Sax
>Assignee: huxihx
>Priority: Critical
> Labels: flaky-test
> Fix For: 2.3.0, 2.2.1
>
>
> To get stable nightly builds for `2.2` release, I create tickets for all
> observed test failures.
> [https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/21/]
> {quote}java.lang.AssertionError: Received 0, expected at least 68 at
> org.junit.Assert.fail(Assert.java:88) at
> org.junit.Assert.assertTrue(Assert.java:41) at
> kafka.api.ConsumerBounceTest.receiveAndCommit(ConsumerBounceTest.scala:557)
> at
> kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1(ConsumerBounceTest.scala:320)
> at
> kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1$adapted(ConsumerBounceTest.scala:319)
> at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
> at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) at
> kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup(ConsumerBounceTest.scala:319){quote}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-7965) Flaky Test ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
[
https://issues.apache.org/jira/browse/KAFKA-7965?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812335#comment-16812335
]
huxihx edited comment on KAFKA-7965 at 4/8/19 10:51 AM:
Seems [https://github.com/apache/kafka/pull/6238] changed the code a lot. Still
got the AssertionError right now?
was (Author: huxi_2b):
Seems [PR 6328|[https://github.com/apache/kafka/pull/6238]] changed the code a
lot. Still got the AssertionError right now?
> Flaky Test
> ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
>
>
> Key: KAFKA-7965
> URL: https://issues.apache.org/jira/browse/KAFKA-7965
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, unit tests
>Affects Versions: 2.2.0, 2.3.0
>Reporter: Matthias J. Sax
>Assignee: huxihx
>Priority: Critical
> Labels: flaky-test
> Fix For: 2.3.0, 2.2.1
>
>
> To get stable nightly builds for `2.2` release, I create tickets for all
> observed test failures.
> [https://jenkins.confluent.io/job/apache-kafka-test/job/2.2/21/]
> {quote}java.lang.AssertionError: Received 0, expected at least 68 at
> org.junit.Assert.fail(Assert.java:88) at
> org.junit.Assert.assertTrue(Assert.java:41) at
> kafka.api.ConsumerBounceTest.receiveAndCommit(ConsumerBounceTest.scala:557)
> at
> kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1(ConsumerBounceTest.scala:320)
> at
> kafka.api.ConsumerBounceTest.$anonfun$testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup$1$adapted(ConsumerBounceTest.scala:319)
> at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
> at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) at
> kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup(ConsumerBounceTest.scala:319){quote}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7983) supporting replication.throttled.replicas in dynamic broker configuration
[ https://issues.apache.org/jira/browse/KAFKA-7983?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812313#comment-16812313 ] Rahul Agarwal commented on KAFKA-7983: -- Can you please assign this ticket to me? Jira Id: rahul.mnit > supporting replication.throttled.replicas in dynamic broker configuration > - > > Key: KAFKA-7983 > URL: https://issues.apache.org/jira/browse/KAFKA-7983 > Project: Kafka > Issue Type: New Feature > Components: core >Reporter: Jun Rao >Priority: Major > > In > [KIP-226|https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration#KIP-226-DynamicBrokerConfiguration-DefaultTopicconfigs], > we added the support to change broker defaults dynamically. However, it > didn't support changing leader.replication.throttled.replicas and > follower.replication.throttled.replicas. These 2 configs were introduced in > [KIP-73|https://cwiki.apache.org/confluence/display/KAFKA/KIP-73+Replication+Quotas] > and controls the set of topic partitions on which replication throttling > will be engaged. One useful case is to be able to set a default value for > both configs to * to allow throttling to be engaged for all topic partitions. > Currently, the static default value for both configs are ignored for > replication throttling, it would be useful to fix that as well. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8194) MessagesInPerSec incorrect value when Stream produce messages
[
https://issues.apache.org/jira/browse/KAFKA-8194?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812303#comment-16812303
]
Odyldzhon Toshbekov commented on KAFKA-8194:
[~guozhang] I can do it during the week but currently I don't have rights to
create KIP on Confluence. My login : odyldzhon
> MessagesInPerSec incorrect value when Stream produce messages
> -
>
> Key: KAFKA-8194
> URL: https://issues.apache.org/jira/browse/KAFKA-8194
> Project: Kafka
> Issue Type: Bug
> Components: metrics
>Affects Versions: 1.1.0, 2.2.0
>Reporter: Odyldzhon Toshbekov
>Priority: Trivial
> Attachments: Screen Shot 2019-04-05 at 17.51.03.png, Screen Shot
> 2019-04-05 at 17.52.22.png
>
>
> Looks like metric
> {code:java}
> kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec{code}
> has incorrect value when messages come via Kafka Stream API.
> I noticed that offset for every message from Kafka Stream can be increased by
> 1,2,... However if messages come to Broker from Kafka producer it's always
> incremented by 1.
> Unfortunately the metric mentioned above calculated based on offset changes
> and as result we cannot use streams because metric will be always incorrect.
> For Kafka 2.2.0
> !Screen Shot 2019-04-05 at 17.51.03.png|width=100%!
>
> [https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/server/ReplicaManager.scala]
> And this is the method used to get "numAppendedMessages"
> !Screen Shot 2019-04-05 at 17.52.22.png|width=100%!
> https://github.com/apache/kafka/blob/2.2.0/core/src/main/scala/kafka/log/Log.scala
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-4600) Consumer proceeds on when ConsumerRebalanceListener fails
[ https://issues.apache.org/jira/browse/KAFKA-4600?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16812168#comment-16812168 ] Braedon Vickers commented on KAFKA-4600: Hi [~guozhang], Sounds like we are on the same page about the issue itself now, which is great :) Option 1 in your comment seems sufficient to me - it's certainly the simpler option of the two. I'm not up on the meta around when KIPs need to be used, but this seems like a pretty basic (but important) bug fix to me, rather than a breaking change to functionality. Are people _really_ expecting uncaught exceptions thrown by `ConsumerRebalanceListener` methods to be (effectively) ignored, and relying intentionally/specifically on that behaviour for their implementations? I'm sure there will be implementations that are silently failing currently (like mine was before I spotted and worked around the issue) that will fail loudly after this is fixed; all the more reason to patch this as a bug. As noted in the original issue, this can easily cause data corruption or data loss. Braedon > Consumer proceeds on when ConsumerRebalanceListener fails > - > > Key: KAFKA-4600 > URL: https://issues.apache.org/jira/browse/KAFKA-4600 > Project: Kafka > Issue Type: Bug > Components: consumer >Affects Versions: 0.10.1.1 >Reporter: Braedon Vickers >Priority: Major > > One of the use cases for a ConsumerRebalanceListener is to load state > necessary for processing a partition when it is assigned. However, when > ConsumerRebalanceListener.onPartitionsAssigned() fails for some reason (i.e. > the state isn't loaded), the error is logged and the consumer proceeds on as > if nothing happened, happily consuming messages from the new partition. When > the state is relied upon for correct processing, this can be very bad, e.g. > data loss can occur. > It would be better if the error was propagated up so it could be dealt with > normally. At the very least the assignment should fail so the consumer > doesn't see any messages from the new partitions, and the rebalance can be > reattempted. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
