[jira] [Commented] (KAFKA-7277) Migrate Streams API to Duration instead of longMs times
[ https://issues.apache.org/jira/browse/KAFKA-7277?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586938#comment-16586938 ] Nikolay Izhikov commented on KAFKA-7277: Hello, [~vvcephei]. I've addressed all your comments. This is my first KIP, so, please, give me advice. Should I start a vote now? Or I should wait some other feedback? For how long? > Migrate Streams API to Duration instead of longMs times > --- > > Key: KAFKA-7277 > URL: https://issues.apache.org/jira/browse/KAFKA-7277 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Assignee: Nikolay Izhikov >Priority: Major > Labels: beginner, kip, newbie > > Right now Streams API unversally represents time as ms-since-unix-epoch. > There's nothing wrong, per se, with this, but Duration is more ergonomic for > an API. > What we don't want is to present a heterogeneous API, so we need to make sure > the whole Streams API is in terms of Duration. > > Implementation note: Durations potentially worsen memory pressure and gc > performance, so internally, we will still use longMs as the representation. > KIP: > [https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586927#comment-16586927
]
ASF GitHub Bot commented on KAFKA-7316:
---
tedyu opened a new pull request #5543: KAFKA-7316 Use of filter method in
KTable.scala may result in StackOverflowError
URL: https://github.com/apache/kafka/pull/5543
Due to lack of conversion to kstream Predicate, existing filter method in
KTable.scala would result in StackOverflowError.
This PR fixes the bug and adds calls in
StreamToTableJoinScalaIntegrationTestImplicitSerdes.testShouldCountClicksPerRegion
to prevent regression.
### 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 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
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt, 7316.v2.txt, 7316.v4.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586909#comment-16586909
]
Ted Yu commented on KAFKA-7316:
---
Patch v4 makes the code compile.
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt, 7316.v2.txt, 7316.v4.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ted Yu updated KAFKA-7316:
--
Attachment: 7316.v4.txt
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt, 7316.v2.txt, 7316.v4.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7190) Under low traffic conditions purging repartition topics cause WARN statements about UNKNOWN_PRODUCER_ID
[ https://issues.apache.org/jira/browse/KAFKA-7190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586903#comment-16586903 ] lambdaliu commented on KAFKA-7190: -- Hi [~mjsax], Changing the log level to DEBUG can't avoid the UNKNOWN_PRODUCER error which can cause one more Request-Response to reset producer epoch and sequence. > Under low traffic conditions purging repartition topics cause WARN statements > about UNKNOWN_PRODUCER_ID > - > > Key: KAFKA-7190 > URL: https://issues.apache.org/jira/browse/KAFKA-7190 > Project: Kafka > Issue Type: Improvement > Components: core, streams >Affects Versions: 1.1.0, 1.1.1 >Reporter: Bill Bejeck >Assignee: lambdaliu >Priority: Major > > When a streams application has little traffic, then it is possible that > consumer purging would delete > even the last message sent by a producer (i.e., all the messages sent by > this producer have been consumed and committed), and as a result, the broker > would delete that producer's ID. The next time when this producer tries to > send, it will get this UNKNOWN_PRODUCER_ID error code, but in this case, > this error is retriable: the producer would just get a new producer id and > retries, and then this time it will succeed. > > Possible fixes could be on the broker side, i.e., delaying the deletion of > the produderIDs for a more extended period or on the streams side developing > a more conservative approach to deleting offsets from repartition topics > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586897#comment-16586897
]
Ted Yu commented on KAFKA-7316:
---
Patch v2 reduces compilation errors to 20.
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt, 7316.v2.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ted Yu updated KAFKA-7316:
--
Attachment: 7316.v2.txt
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt, 7316.v2.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7190) Under low traffic conditions purging repartition topics cause WARN statements about UNKNOWN_PRODUCER_ID
[ https://issues.apache.org/jira/browse/KAFKA-7190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586886#comment-16586886 ] lambdaliu commented on KAFKA-7190: -- Hi [~hachikuji], sorry for the later reply. I have thought about the solution you suggested last time, and found it's not easy to reset sequence to 0 for each transaction. Because the broker may receive a new transactional produce request before the last complete transaction's EndTxnMaker request. So we would better cache all PID in memory until it expired. For the idea of soft delete records which beyond the LSO, it's easy to implement. But the deletion of retention time and retention size breach may still delete the segment which contain active transaction. For this case, maybe we can use snapshot file to save the active PID. With this change we can always recover PID which have active transaction from log. For the KIP you posted seems great and I am glad to work on it. Thanks. > Under low traffic conditions purging repartition topics cause WARN statements > about UNKNOWN_PRODUCER_ID > - > > Key: KAFKA-7190 > URL: https://issues.apache.org/jira/browse/KAFKA-7190 > Project: Kafka > Issue Type: Improvement > Components: core, streams >Affects Versions: 1.1.0, 1.1.1 >Reporter: Bill Bejeck >Assignee: lambdaliu >Priority: Major > > When a streams application has little traffic, then it is possible that > consumer purging would delete > even the last message sent by a producer (i.e., all the messages sent by > this producer have been consumed and committed), and as a result, the broker > would delete that producer's ID. The next time when this producer tries to > send, it will get this UNKNOWN_PRODUCER_ID error code, but in this case, > this error is retriable: the producer would just get a new producer id and > retries, and then this time it will succeed. > > Possible fixes could be on the broker side, i.e., delaying the deletion of > the produderIDs for a more extended period or on the streams side developing > a more conservative approach to deleting offsets from repartition topics > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586841#comment-16586841
]
Ted Yu commented on KAFKA-7316:
---
[~guozhang] [~mjsax] :
Can you take a look at patch v1 to see if the changes to
SessionWindowedKStream.scala are acceptable ?
If so, I can work through the rest of compilation errors.
Thanks
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ted Yu updated KAFKA-7316:
--
Attachment: 7316.v1.txt
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
> Attachments: 7316.v1.txt
>
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586738#comment-16586738
]
Ted Yu commented on KAFKA-7316:
---
I was thinking about removing the implicit wrapKTable .
Then the following compilation errors pop up (only a snippet, there are more):
{code}
/Users/tyu/kafka/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala:52:
type mismatch;
found :
org.apache.kafka.streams.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],VR]
required:
org.apache.kafka.streams.scala.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],VR]
inner.aggregate((() => initializer).asInitializer, aggregator.asAggregator,
merger.asMerger, materialized)
^
/Users/tyu/kafka/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala:64:
type mismatch;
found :
org.apache.kafka.streams.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],Long]
required:
org.apache.kafka.streams.scala.kstream.KTable[org.apache.kafka.streams.kstream.Windowed[K],Long]
inner.count(materialized.asInstanceOf[Materialized[K, java.lang.Long,
ByteArraySessionStore]])
{code}
If modifying the individual places is acceptable, I can send a PR.
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586709#comment-16586709 ] Matthias J. Sax commented on KAFKA-7315: `Serde` are a Streams concept – it's an abstraction that wraps a serializer and a deserializer in a single class. We can of course add serialization and deserialization docs for producer and consumer respectively. No sure, if a centralized place that explains all three things for the 3 APIs would make sense. If we document serialization/deserialization within Consumer and Producer docs (or maybe have a section "data types" for both), the `Serde` datatype section in Streams can just link to both, and only explain how to create corresponding Serdes using the plain Serializer/Deserializer (similar to what we have for JSON already). Adding other formats than AVRO makes sense to me in principle – however, we should be careful to not blow up the docs – there are many hundreds of data formats out, and we don't want to document all of them. As you pointed out, Kafka uses byte[] format internally and there is no good reason to add unnecessary docs on "how to (de)serialize datatype X". The question we should answer is, do we add value adding those docs, or do we duplicate things that are documented somewhere else? So don't get me wrong. I am always in favor of improving/extending the docs in general. We should just think/discuss about what makes sense and what not. Looking forward to the opinion of others. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-7316:
---
Affects Version/s: 2.0.0
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-7316:
---
Labels: scala (was: )
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
[
https://issues.apache.org/jira/browse/KAFKA-7316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-7316:
---
Component/s: streams
> Use of filter method in KTable.scala may result in StackOverflowError
> -
>
> Key: KAFKA-7316
> URL: https://issues.apache.org/jira/browse/KAFKA-7316
> Project: Kafka
> Issue Type: Bug
> Components: streams
>Affects Versions: 2.0.0
>Reporter: Ted Yu
>Priority: Major
> Labels: scala
>
> In this thread:
> http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
> Druhin reported seeing StackOverflowError when using filter method from
> KTable.scala
> This can be reproduced with the following change:
> {code}
> diff --git
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> b/streams/streams-scala/src/test/scala
> index 3d1bab5..e0a06f2 100644
> ---
> a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> +++
> b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> @@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
> extends StreamToTableJ
> val userClicksStream: KStream[String, Long] =
> builder.stream(userClicksTopic)
> val userRegionsTable: KTable[String, String] =
> builder.table(userRegionsTopic)
> +userRegionsTable.filter { case (_, count) => true }
> // Compute the total per region by summing the individual click counts
> per region.
> val clicksPerRegion: KTable[String, Long] =
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-7316) Use of filter method in KTable.scala may result in StackOverflowError
Ted Yu created KAFKA-7316:
-
Summary: Use of filter method in KTable.scala may result in
StackOverflowError
Key: KAFKA-7316
URL: https://issues.apache.org/jira/browse/KAFKA-7316
Project: Kafka
Issue Type: Bug
Reporter: Ted Yu
In this thread:
http://search-hadoop.com/m/Kafka/uyzND1dNbYKXzC4F1?subj=Issue+in+Kafka+2+0+0+
Druhin reported seeing StackOverflowError when using filter method from
KTable.scala
This can be reproduced with the following change:
{code}
diff --git
a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
b/streams/streams-scala/src/test/scala
index 3d1bab5..e0a06f2 100644
---
a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
+++
b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
@@ -58,6 +58,7 @@ class StreamToTableJoinScalaIntegrationTestImplicitSerdes
extends StreamToTableJ
val userClicksStream: KStream[String, Long] =
builder.stream(userClicksTopic)
val userRegionsTable: KTable[String, String] =
builder.table(userRegionsTopic)
+userRegionsTable.filter { case (_, count) => true }
// Compute the total per region by summing the individual click counts per
region.
val clicksPerRegion: KTable[String, Long] =
{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-5891) Cast transformation fails if record schema contains timestamp field
[
https://issues.apache.org/jira/browse/KAFKA-5891?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jason Gustafson resolved KAFKA-5891.
Resolution: Fixed
Fix Version/s: 2.0.1
1.1.2
1.0.3
> Cast transformation fails if record schema contains timestamp field
> ---
>
> Key: KAFKA-5891
> URL: https://issues.apache.org/jira/browse/KAFKA-5891
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 0.11.0.0
>Reporter: Artem Plotnikov
>Priority: Major
> Fix For: 1.0.3, 1.1.2, 2.0.1
>
>
> I have the following simple type cast transformation:
> {code}
> name=postgresql-source-simple
> connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
> tasks.max=1
> connection.url=jdbc:postgresql://localhost:5432/testdb?user=postgres=mysecretpassword
> query=SELECT 1::INT as a, '2017-09-14 10:23:54'::TIMESTAMP as b
> transforms=Cast
> transforms.Cast.type=org.apache.kafka.connect.transforms.Cast$Value
> transforms.Cast.spec=a:boolean
> mode=bulk
> topic.prefix=clients
> {code}
> Which fails with the following exception in runtime:
> {code}
> [2017-09-14 16:51:01,885] ERROR Task postgresql-source-simple-0 threw an
> uncaught and unrecoverable exception
> (org.apache.kafka.connect.runtime.WorkerTask:148)
> org.apache.kafka.connect.errors.DataException: Invalid Java object for schema
> type INT64: class java.sql.Timestamp for field: "null"
> at
> org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:239)
> at
> org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:209)
> at org.apache.kafka.connect.data.Struct.put(Struct.java:214)
> at
> org.apache.kafka.connect.transforms.Cast.applyWithSchema(Cast.java:152)
> at org.apache.kafka.connect.transforms.Cast.apply(Cast.java:108)
> at
> org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38)
> at
> org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:190)
> at
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:168)
> at
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
> at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {code}
> If I remove the transforms.* part of the connector it will work correctly.
> Actually, it doesn't really matter which types I use in the transformation
> for field 'a', just the existence of a timestamp field brings the exception.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-5891) Cast transformation fails if record schema contains timestamp field
[
https://issues.apache.org/jira/browse/KAFKA-5891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586665#comment-16586665
]
ASF GitHub Bot commented on KAFKA-5891:
---
hachikuji closed pull request #4633: KAFKA-5891: Proper handling of
LogicalTypes in Cast
URL: https://github.com/apache/kafka/pull/4633
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
index d94f8f648b3..22b19722c47 100644
---
a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
+++
b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
@@ -164,13 +164,17 @@ private Schema getOrBuildSchema(Schema valueSchema) {
} else {
builder = SchemaUtil.copySchemaBasics(valueSchema,
SchemaBuilder.struct());
for (Field field : valueSchema.fields()) {
-SchemaBuilder fieldBuilder =
-convertFieldType(casts.containsKey(field.name()) ?
casts.get(field.name()) : field.schema().type());
-if (field.schema().isOptional())
-fieldBuilder.optional();
-if (field.schema().defaultValue() != null)
-
fieldBuilder.defaultValue(castValueToType(field.schema().defaultValue(),
fieldBuilder.type()));
-builder.field(field.name(), fieldBuilder.build());
+if (casts.containsKey(field.name())) {
+SchemaBuilder fieldBuilder =
convertFieldType(casts.get(field.name()));
+if (field.schema().isOptional())
+fieldBuilder.optional();
+if (field.schema().defaultValue() != null)
+
fieldBuilder.defaultValue(castValueToType(field.schema().defaultValue(),
fieldBuilder.type()));
+builder.field(field.name(), fieldBuilder.build());
+} else {
+builder.field(field.name(), field.schema());
+}
+
}
}
diff --git
a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
index b190189b35d..decd043b1db 100644
---
a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
+++
b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
@@ -21,12 +21,14 @@
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.After;
import org.junit.Test;
import java.util.Collections;
+import java.util.Date;
import java.util.HashMap;
import java.util.Map;
@@ -304,6 +306,7 @@ public void castFieldsWithSchema() {
builder.field("boolean", Schema.BOOLEAN_SCHEMA);
builder.field("string", Schema.STRING_SCHEMA);
builder.field("optional", Schema.OPTIONAL_FLOAT32_SCHEMA);
+builder.field("timestamp", Timestamp.SCHEMA);
Schema supportedTypesSchema = builder.build();
Struct recordValue = new Struct(supportedTypesSchema);
@@ -315,6 +318,7 @@ public void castFieldsWithSchema() {
recordValue.put("float64", -64.);
recordValue.put("boolean", true);
recordValue.put("string", "42");
+recordValue.put("timestamp", new Date(0));
// optional field intentionally omitted
SourceRecord transformed = xformValue.apply(new SourceRecord(null,
null, "topic", 0,
@@ -331,6 +335,7 @@ public void castFieldsWithSchema() {
assertEquals(true, ((Struct)
transformed.value()).schema().field("float64").schema().defaultValue());
assertEquals((byte) 1, ((Struct) transformed.value()).get("boolean"));
assertEquals(42, ((Struct) transformed.value()).get("string"));
+assertEquals(new Date(0), ((Struct)
transformed.value()).get("timestamp"));
assertNull(((Struct) transformed.value()).get("optional"));
}
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
> Cast transformation fails if record schema contains
[jira] [Comment Edited] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586650#comment-16586650 ] Jordan Moore edited comment on KAFKA-7315 at 8/20/18 11:12 PM: --- What about a separate section on SerDe's outside of the Streams docs? I've seen examples of people using MsgPack and ProtoBuf as well. Sure, it's documented implicitly (or briefly under "Message Format") that Kafka uses only byte[] for keys/values, but lots of "getting started guides" are sending strings and not really discussing the benefits of alternative, more compact "structured" formats. was (Author: cricket007): What about a separate section on SerDe's outside of the Streams docs? I've seen examples of people using MsgPack and ProtoBuf as well. Sure, it's documented implicitly (or briefly under "Message Format") that Kafka uses only byte[] for keys/values, but lots of "getting started guides" are sending strings and not really discussing the values of alternative, more compact "structured" formats. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586650#comment-16586650 ] Jordan Moore commented on KAFKA-7315: - What about a separate section on SerDe's outside of the Streams docs? I've seen examples of people using MsgPack and ProtoBuf as well. Sure, it's documented implicitly (or briefly under "Message Format") that Kafka uses only byte[] for keys/values, but lots of "getting started guides" are sending strings and not really discussing the values of alternative, more compact "structured" formats. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-6961) UnknownTopicOrPartitionException & NotLeaderForPartitionException upon replication of topics.
[ https://issues.apache.org/jira/browse/KAFKA-6961?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586615#comment-16586615 ] Jiangtao Liu edited comment on KAFKA-6961 at 8/20/18 11:08 PM: --- I encounter the same issue after I move Kafka server to new EC2 server. so any idea to solve it? or simply ignore it since our replica is functioning well. was (Author: tony2011): I encounter the same issue after I move Kafka server to new EC2 server. so any idea to solve it? or simply ignore it since our replica is functional well. > UnknownTopicOrPartitionException & NotLeaderForPartitionException upon > replication of topics. > - > > Key: KAFKA-6961 > URL: https://issues.apache.org/jira/browse/KAFKA-6961 > Project: Kafka > Issue Type: Bug > Environment: kubernetes cluster kafka. >Reporter: kaushik srinivas >Assignee: praveen >Priority: Major > Attachments: k8s_NotLeaderForPartition.txt, k8s_replication_errors.txt > > > Running kafka & zookeeper in kubernetes cluster. > No of brokers : 3 > No of partitions per topic : 3 > creating topic with 3 partitions, and looks like all the partitions are up. > Below is the snapshot to confirm the same, > Topic:applestore PartitionCount:3 ReplicationFactor:3 Configs: > Topic: applestore Partition: 0Leader: 1001Replicas: > 1001,1003,1002Isr: 1001,1003,1002 > Topic: applestore Partition: 1Leader: 1002Replicas: > 1002,1001,1003Isr: 1002,1001,1003 > Topic: applestore Partition: 2Leader: 1003Replicas: > 1003,1002,1001Isr: 1003,1002,1001 > > But, we see in the brokers as soon as the topics are created below stack > traces appears, > > error 1: > [2018-05-28 08:00:31,875] ERROR [ReplicaFetcher replicaId=1001, > leaderId=1003, fetcherId=7] Error for partition applestore-2 to broker > 1003:org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This > server does not host this topic-partition. (kafka.server.ReplicaFetcherThread) > > error 2 : > [2018-05-28 00:43:20,993] ERROR [ReplicaFetcher replicaId=1003, > leaderId=1001, fetcherId=0] Error for partition apple-6 to broker > 1001:org.apache.kafka.common.errors.NotLeaderForPartitionException: This > server is not the leader for that topic-partition. > (kafka.server.ReplicaFetcherThread) > > When we tries producing records to each specific partition, it works fine and > also log size across the replicated brokers appears to be equal, which means > replication is happening fine. > Attaching the two stack trace files. > > Why are these stack traces appearing ? can we ignore these stack traces if > its some spam messages ? > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586635#comment-16586635 ] Matthias J. Sax commented on KAFKA-7315: It's just my personal thought :) (I like easy fixes). We can of course add an AVRO section, too. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Comment Edited] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586629#comment-16586629 ] Jordan Moore edited comment on KAFKA-7315 at 8/20/18 10:56 PM: --- bq. should be vendor agnostic and not mention any of them Fair enough. Though, in its place I feel it would be fair to mention at least _somewhere_ the alternatives to primitive Java types, byte arrays and JSON. The only mentions of Avro elsewhere in the docs that I can easily find are in the Connect sections. was (Author: cricket007): bq. should be vendor agnostic and not mention any of them Fair enough. Though, in it's place I feel it would be fair to mention at least _somewhere_ the alternatives to primitive Java types, byte arrays and JSON. The only mentions of Avro elsewhere in the docs that I can easily find are in the Connect sections. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586629#comment-16586629 ] Jordan Moore commented on KAFKA-7315: - bq. should be vendor agnostic and not mention any of them Fair enough. Though, in it's place I feel it would be fair to mention at least _somewhere_ the alternatives to primitive Java types, byte arrays and JSON. The only mentions of Avro elsewhere in the docs that I can easily find are in the Connect sections. > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-6961) UnknownTopicOrPartitionException & NotLeaderForPartitionException upon replication of topics.
[ https://issues.apache.org/jira/browse/KAFKA-6961?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586615#comment-16586615 ] Jiangtao Liu commented on KAFKA-6961: - I encounter the same issue after I move Kafka server to new EC2 server. so any idea to solve it? or simply ignore it since our replica is functional well. > UnknownTopicOrPartitionException & NotLeaderForPartitionException upon > replication of topics. > - > > Key: KAFKA-6961 > URL: https://issues.apache.org/jira/browse/KAFKA-6961 > Project: Kafka > Issue Type: Bug > Environment: kubernetes cluster kafka. >Reporter: kaushik srinivas >Assignee: praveen >Priority: Major > Attachments: k8s_NotLeaderForPartition.txt, k8s_replication_errors.txt > > > Running kafka & zookeeper in kubernetes cluster. > No of brokers : 3 > No of partitions per topic : 3 > creating topic with 3 partitions, and looks like all the partitions are up. > Below is the snapshot to confirm the same, > Topic:applestore PartitionCount:3 ReplicationFactor:3 Configs: > Topic: applestore Partition: 0Leader: 1001Replicas: > 1001,1003,1002Isr: 1001,1003,1002 > Topic: applestore Partition: 1Leader: 1002Replicas: > 1002,1001,1003Isr: 1002,1001,1003 > Topic: applestore Partition: 2Leader: 1003Replicas: > 1003,1002,1001Isr: 1003,1002,1001 > > But, we see in the brokers as soon as the topics are created below stack > traces appears, > > error 1: > [2018-05-28 08:00:31,875] ERROR [ReplicaFetcher replicaId=1001, > leaderId=1003, fetcherId=7] Error for partition applestore-2 to broker > 1003:org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This > server does not host this topic-partition. (kafka.server.ReplicaFetcherThread) > > error 2 : > [2018-05-28 00:43:20,993] ERROR [ReplicaFetcher replicaId=1003, > leaderId=1001, fetcherId=0] Error for partition apple-6 to broker > 1001:org.apache.kafka.common.errors.NotLeaderForPartitionException: This > server is not the leader for that topic-partition. > (kafka.server.ReplicaFetcherThread) > > When we tries producing records to each specific partition, it works fine and > also log size across the replicated brokers appears to be equal, which means > replication is happening fine. > Attaching the two stack trace files. > > Why are these stack traces appearing ? can we ignore these stack traces if > its some spam messages ? > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6149) LogCleanerManager should include topic partition name when warning of invalid cleaner offset
[ https://issues.apache.org/jira/browse/KAFKA-6149?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6149: -- Component/s: log > LogCleanerManager should include topic partition name when warning of invalid > cleaner offset > - > > Key: KAFKA-6149 > URL: https://issues.apache.org/jira/browse/KAFKA-6149 > Project: Kafka > Issue Type: Improvement > Components: log, logging >Reporter: Ryan P >Priority: Major > > The following message would be a lot more helpful if the topic partition name > were included. > if (!isCompactAndDelete(log)) > warn(s"Resetting first dirty offset to log start offset > $logStartOffset since the checkpointed offset $offset is invalid.") -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-5943) Reduce dependency on mock in connector tests
[ https://issues.apache.org/jira/browse/KAFKA-5943?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-5943: -- Component/s: unit tests > Reduce dependency on mock in connector tests > > > Key: KAFKA-5943 > URL: https://issues.apache.org/jira/browse/KAFKA-5943 > Project: Kafka > Issue Type: Test > Components: unit tests >Reporter: Ted Yu >Priority: Minor > Labels: connector, mock > > Currently connector tests make heavy use of mock (easymock, power mock). > This may hide the real logic behind operations and makes finding bugs > difficult. > We should reduce the use of mocks so that developers can debug connector code > using unit tests. > This would shorten the development cycle for connector. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6033) Kafka Streams does not work with musl-libc (alpine linux)
[ https://issues.apache.org/jira/browse/KAFKA-6033?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6033: -- Component/s: streams > Kafka Streams does not work with musl-libc (alpine linux) > - > > Key: KAFKA-6033 > URL: https://issues.apache.org/jira/browse/KAFKA-6033 > Project: Kafka > Issue Type: Improvement > Components: streams >Affects Versions: 0.11.0.0 > Environment: Alpine 3.6 >Reporter: Jeffrey Zampieron >Priority: Major > > Using the released version of kafka fails on alpine based images b/c of > rocksdb using the jni and failing to load. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6149) LogCleanerManager should include topic partition name when warning of invalid cleaner offset
[ https://issues.apache.org/jira/browse/KAFKA-6149?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6149: -- Component/s: logging > LogCleanerManager should include topic partition name when warning of invalid > cleaner offset > - > > Key: KAFKA-6149 > URL: https://issues.apache.org/jira/browse/KAFKA-6149 > Project: Kafka > Issue Type: Improvement > Components: log, logging >Reporter: Ryan P >Priority: Major > > The following message would be a lot more helpful if the topic partition name > were included. > if (!isCompactAndDelete(log)) > warn(s"Resetting first dirty offset to log start offset > $logStartOffset since the checkpointed offset $offset is invalid.") -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6295) Add 'Coordinator Id' to consumer metrics
[ https://issues.apache.org/jira/browse/KAFKA-6295?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6295: -- Component/s: metrics > Add 'Coordinator Id' to consumer metrics > > > Key: KAFKA-6295 > URL: https://issues.apache.org/jira/browse/KAFKA-6295 > Project: Kafka > Issue Type: Improvement > Components: metrics >Reporter: Ryan P >Priority: Major > Labels: needs-kip > > It would be incredibly helpful to be able to review which broker was the > coordinator for a consumer at a given point in time. The easiest way to > achieve this in my opinion would be to expose a coordinator id metric. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6143) VerifiableProducer & VerifiableConsumer need tests
[
https://issues.apache.org/jira/browse/KAFKA-6143?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-6143:
--
Component/s: system tests
> VerifiableProducer & VerifiableConsumer need tests
> --
>
> Key: KAFKA-6143
> URL: https://issues.apache.org/jira/browse/KAFKA-6143
> Project: Kafka
> Issue Type: Bug
> Components: system tests
>Reporter: Tom Bentley
>Priority: Minor
>
> The {{VerifiableProducer}} and {{VerifiableConsumer}} used use for system
> tests, but don't have any tests themselves. They should have.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-6385) Rack awareness ignored by kafka-reassign-partitions
[
https://issues.apache.org/jira/browse/KAFKA-6385?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-6385:
--
Component/s: tools
> Rack awareness ignored by kafka-reassign-partitions
> ---
>
> Key: KAFKA-6385
> URL: https://issues.apache.org/jira/browse/KAFKA-6385
> Project: Kafka
> Issue Type: Bug
> Components: tools
>Affects Versions: 1.0.0
> Environment: Ubuntu 16.04
>Reporter: Gal Barak
>Priority: Minor
> Attachments: actual.txt, topic-to-move.json
>
>
> Hi,
> It seems that the kafka-reassign-partitions script ignores rack awareness,
> when suggesting a new partition layout. Came across it when doing some
> initial testing with Kafka.
> +To reproduce:+
> # Create a Kafka cluster with 3 brokers (1,2,3). Use 3 different racks
> (broker.rack definition. Example: "A", "B" and "C").
> #* I used a non-root directory in zookeeper (i.e. - {{ 1>:2181,:2181,:2182/}})
> #* The tested topic was automatically created, according to a default
> configuration of 12 partitions and 3 replicas per topic.
> # Install a 4th broker, and assign it to the same rack as the 1st broker
> ("A").
> # Create a topics-to-move.json file for a single topic. The file I used was
> uploaded as topic-to-move.json.
> # Run the kafka-reassign-partitions script:
> {{kafka-reassign-partitions --zookeeper :2181, 2>:2181,:2182/
> --topics-to-move-json-file --broker-list "1,2,3,4"
> --generate}}
> +Expected result:+
> A suggested reassignment that makes sure that no partitions uses both broker
> 1 and broker 4 as its replicas.
> +Actual results of the command:+
> The full result is attached as a file (actual.txt). It includes partitions
> with replicas that are on both brokers 1 and 4, which are two servers on the
> same rack.
> Example: {"topic":"","partition":6,"replicas":[1,2,4]}
> +Additional notes:+
> * I did not test starting the cluster from scratch. The same behavior might
> be present when topic partitions are created automatically (in which case,
> the priority might be higher).
> * I'm not sure it's related. But the original assignment seems to be
> problematic as well: If a single server (of the 3) failed, a different single
> server became the leader for all of its partitions. For example, if broker 1
> failed, server 2 became the leader for all of the partitions for which 1 was
> previously the leader, instead of having the load distributed evenly between
> brokers 2 and 3.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-6154) Transient failure TransactionsBounceTest.testBrokerFailure
[
https://issues.apache.org/jira/browse/KAFKA-6154?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-6154:
--
Component/s: unit tests
> Transient failure TransactionsBounceTest.testBrokerFailure
> --
>
> Key: KAFKA-6154
> URL: https://issues.apache.org/jira/browse/KAFKA-6154
> Project: Kafka
> Issue Type: Bug
> Components: unit tests
>Reporter: Jason Gustafson
>Priority: Major
>
> {code}
> java.lang.AssertionError: Out of order messages detected
> expected: 34, 37, 43, 44, 45, 47, 48, 52, 54, 55, 56, 59, 60, 61, 63, 64, 70, 71, 72,
> 76, 79, 80, 82, 88, 89, 92, 93, 96, 98, 99, 100, 102, 106, 107, 108, 109,
> 119, 125, 131, 132, 134, 137, 141, 144, 145, 153, 154, 156, 157, 158, 160,
> 163, 168, 174, 177, 178, 183, 184, 187, 190, 191, 194, 195, 196, 198, 199,
> 204, 206, 207, 208, 210, 215, 219, 222, 224, 227, 228, 238, 240, 244, 247,
> 255, 256, 257, 261, 265, 267, 269, 272, 273, 275, 278, 279, 281, 282, 283,
> 287, 288, 293, 295, 296, 299, 300, 301, 304, 307, 308, 311, 313, 316, 317,
> 319, 320, 326, 329, 335, 337, 338, 339, 340, 342, 344, 347, 348, 355, 356,
> 360, 361, 364, 366, 367, 369, 371, 373, 389, 391, 395, 398, 402, 404, 405,
> 408, 409, 410, 411, 420, 421, 423, 424, 428, 432, 433, 442, 445, 449, 450,
> 451, 452, 453, 454, 457, 460, 461, 463, 464, 470, 471, 472, 474, 476, 477,
> 483, 485, 489, 491, 492, 493, 494, 499, 500, 501, 502, 505, 507, 510, 515,
> 516, 522, 523, 525, 527, 528, 530, 532, 538, 542, 544, 550, 556, 557, 560,
> 563, 564, 565, 568, 569, 570, 573, 575, 576, 582, 587, 594, 597, 602, 603,
> 605, 608, 609, 610, 613, 621, 623, 631, 632, 638, 642, 644, 651, 654, 660,
> 662, 664, 666, 667, 668, 669, 672, 675, 676, 682, 683, 686, 687, 688, 691,
> 692, 698, 701, 709, 711, 716, 718, 719, 722, 725, 727, 729, 731, 734, 738,
> 739, 740, 743, 744, 745, 751, 758, 762, 764, 767, 771, 773, 775, 781, 783,
> 784, 788, 795, 796, 802, 804, 805, 809, 811, 812, 819, 822, 824, 826, 828,
> 831, 834, 839, 840, 844, 845, 849, 856, 859, 862, 864, 865, 870, 871, 889,
> 895, 899, 900, 901, 907, 913, 916, 921, 922, 924, 925, 931, 938, 941, 945,
> 948, 953, 954, 956, 959, 960, 962, 966, 968, 969, 970, 974, 985, 998, 999,
> 1001, 1002, 1007, 1008, 1015, 1017, 1020, 1022, 1026, 1027, 1028, 1030, 1033,
> 1034, 1036, 1043, 1044, 1046, 1049, 1051, 1054, 1055, 1059, 1060, 1062, 1063,
> 1065, 1069, 1074, 1077, 1079, 1084, 1089, 1090, 1094, 1095, 1097, 1105, 1110,
> 1112, 1113, 1119, 1121, 1126, 1130, 1134, 1135, 1137, 1140, 1141, 1149, 1150,
> 1151, 1154, 1156, 1157, 1159, 1160, 1161, 1162, 1167, 1169, 1172, 1173, 1176,
> 1179, 1184, 1186, 1188, 1195, 1197, 1198, 1199, 1201, 1203, 1205, 1207, 1216,
> 1217, 1219, 1221, 1223, 1226, 1233, 1234, 1236, 1239, 1240, 1241, 1242, 1248,
> 1255, 1256, 1264, 1270, 1271, 1274, 1279, 1283, 1289, 1292, 1293, 1294, 1301,
> 1306, 1307, 1310, 1312, 1313, 1316, 1317, 1322, 1330, 1332, 1333, 1334, 1335,
> 1337, 1342, 1345, 1348, 1351, 1354, 1355, 1357, 1358, 1359, 1363, 1365, 1376,
> 1378, 1381, 1388, 1390, 1391, 1396, 1397, 1398, 1404, 1412, 1413, 1414, 1417,
> 1420, 1423, 1424, 1429, 1434, 1436, 1438, 1443, 1445, 1446, 1448, 1449, 1453,
> 1455, 1461, 1464, 1473, 1474, 1475, 1478, 1479, 1481, 1482, 1484, 1494, 1497,
> 1501, 1505, 1509, 1519, 1520, 1523, 1525, 1529, 1530, 1533, 1534, 1538, 1546,
> 1550, 1551, 1552, 1554, 1555, 1557, 1562, 1565, 1566, 1574, 1577, 1580, 1581,
> 1585, 1586, 1587, 1593, 1600, 1604, 1606, 1609, 1611, 1615, 1618, 1620, 1621,
> 1622, 1623, 1626, 1628, 1629, 1632, 1641, 1643, 1645, 1646, 1647, 1651, 1652,
> 1655, 1662, 1677, 1679, 1680, 1681, 1683, 1684, 1688, 1693, 1695, 1696, 1700,
> 1712, 1718, 1719, 1722, 1723, 1731, 1734, 1735, 1739, 1745, 1746, 1749, 1753,
> 1755, 1756, 1758, 1760, 1761, 1762, 1763, 1765, 1766, 1767, 1768, 1770, 1771,
> 1776, 1777, 1779, 1780, 1782, 1785, 1786, 1793, 1802, 1803, 1804, 1805, 1806,
> 1807, 1811, 1815, 1817, 1819, 1821, 1826, 1830, 1840, 1845, 1846, 1847, 1848,
> 1850, 1855, 1857, 1859, 1862, 1869, 1870, 1876, 1879, 1881, 1882, 1885, 1890,
> 1893, 1895, 1898, 1899, 1900, 1901, 1902, 1903, 1906, 1908, 1911, 1915, 1917,
> 1918, 1919, 1921, 1926, 1927, 1931, 1934, 1936, 1941, 1951, 1952, 1953, 1954,
> 1958, 1961, 1964, 1965, 1966, 1970, 1971, 1972, 1974, 1985, 1987, 1992, 1995,
> 1997, 1999, 2001, 2002, 2005, 2006, 2008, 2009, 2014, 2015, 2019, 2023, 2026,
> 2039, 2040, 2042, 2043, 2050, 2051, 2052, 2053, 2055, 2056, 2057, 2060, 2067,
> 2069, 2070, 2076, 2077, 2083, 2084, 2085, 2086, 2087, 2091, 2093, 2097, 2100,
> 2101, 2102, 2104, 2106, 2107, 2110, 2115, 2117, 2121, 2123, 2130, 2136, 2137,
> 2139, 2148, 2149, 2153, 2156, 2157, 2160, 2163, 2164, 2172, 2179, 2180, 2181,
> 2183, 2184, 2186, 2188, 2189, 2193, 2198, 2199, 2200, 2212, 2213, 2218, ,
>
[jira] [Updated] (KAFKA-5887) Enable findBugs (or equivalent) when building with Java 9
[ https://issues.apache.org/jira/browse/KAFKA-5887?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-5887: -- Component/s: build > Enable findBugs (or equivalent) when building with Java 9 > - > > Key: KAFKA-5887 > URL: https://issues.apache.org/jira/browse/KAFKA-5887 > Project: Kafka > Issue Type: Task > Components: build >Reporter: Ismael Juma >Priority: Major > Fix For: 2.1.0 > > > findBugs doesn't support Java 9 and it seems to be abandonware at this point: > https://github.com/findbugsproject/findbugs/issues/105 > https://github.com/gradle/gradle/issues/720 > It has been forked, but the fork requires Java 8: > https://github.com/spotbugs/spotbugs > https://github.com/spotbugs/spotbugs/blob/master/docs/migration.rst#findbugs-gradle-plugin > We should migrate once we move to Java 8 if spotbugs is still being actively > developed and findBugs continues to be dead. > Additional tasks: > 1. Remove the code that disables the Gradle plugin for findBugs (or spotbugs) > when building with Java 9. > 2. Enable the findBugs plugin in Jenkins for the relevant builds: > https://builds.apache.org/job/kafka-trunk-jdk9/ > https://builds.apache.org/job/kafka-pr-jdk9-scala2.12/ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7202) Support multiple auto-generated docs formats
[ https://issues.apache.org/jira/browse/KAFKA-7202?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7202: -- Component/s: documentation > Support multiple auto-generated docs formats > > > Key: KAFKA-7202 > URL: https://issues.apache.org/jira/browse/KAFKA-7202 > Project: Kafka > Issue Type: New Feature > Components: documentation >Reporter: Joel Hamill >Priority: Major > > Currently the configuration parameters for Confluent/Kafka are autogenerated > as HTML (and hosted at > [https://kafka.apache.org/documentation/#configuration]). This request is to > expand this to support other formats (e.g. RST) so that they can be easily > leveraged by other authorign language formats. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6823) Transient failure in DynamicBrokerReconfigurationTest.testThreadPoolResize
[ https://issues.apache.org/jira/browse/KAFKA-6823?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6823: -- Component/s: unit tests > Transient failure in DynamicBrokerReconfigurationTest.testThreadPoolResize > -- > > Key: KAFKA-6823 > URL: https://issues.apache.org/jira/browse/KAFKA-6823 > Project: Kafka > Issue Type: Bug > Components: unit tests >Reporter: Anna Povzner >Priority: Major > > Saw in my PR build (*DK 10 and Scala 2.12 ):* > *15:58:46* kafka.server.DynamicBrokerReconfigurationTest > > testThreadPoolResize FAILED > *15:58:46* java.lang.AssertionError: Invalid threads: expected 6, got 7: > List(ReplicaFetcherThread-0-0, ReplicaFetcherThread-0-1, > ReplicaFetcherThread-0-2, ReplicaFetcherThread-0-0, ReplicaFetcherThread-0-2, > ReplicaFetcherThread-1-0, ReplicaFetcherThread-0-1) > *15:58:46* at org.junit.Assert.fail(Assert.java:88) > *15:58:46* at org.junit.Assert.assertTrue(Assert.java:41) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.verifyThreads(DynamicBrokerReconfigurationTest.scala:1147) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.maybeVerifyThreadPoolSize$1(DynamicBrokerReconfigurationTest.scala:412) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.resizeThreadPool$1(DynamicBrokerReconfigurationTest.scala:431) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.reducePoolSize$1(DynamicBrokerReconfigurationTest.scala:417) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.$anonfun$testThreadPoolResize$3(DynamicBrokerReconfigurationTest.scala:440) > *15:58:46* at > scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:156) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.verifyThreadPoolResize$1(DynamicBrokerReconfigurationTest.scala:439) > *15:58:46* at > kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize(DynamicBrokerReconfigurationTest.scala:453) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-6465) Add a metrics for the number of records per log
[ https://issues.apache.org/jira/browse/KAFKA-6465?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6465: -- Component/s: log > Add a metrics for the number of records per log > --- > > Key: KAFKA-6465 > URL: https://issues.apache.org/jira/browse/KAFKA-6465 > Project: Kafka > Issue Type: Bug > Components: log >Reporter: Ivan Babrou >Priority: Major > > Currently there are log metrics for: > * Start offset > * End offset > * Size in bytes > * Number of segments > I propose to add another metric to track number of record batches in the log. > This should provide operators with an idea of how much batching is happening > on the producers. Having this metric in one place seems easier than scraping > the metric from each producer. > Having an absolute counter may be infeasible (batches are not assigned > sequential IDs), but gauge should be ok. Average batch size can be calculated > as (end offset - start offset) / number of batches. This will be heavily > skewed for logs with long retention, though. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7166) Links are really difficult to see - insufficient contrast
[
https://issues.apache.org/jira/browse/KAFKA-7166?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-7166:
--
Component/s: documentation
> Links are really difficult to see - insufficient contrast
> -
>
> Key: KAFKA-7166
> URL: https://issues.apache.org/jira/browse/KAFKA-7166
> Project: Kafka
> Issue Type: Bug
> Components: documentation
>Reporter: Sebb
>Priority: Major
>
> {color:#0b6d88}The link colour on the pages is very similar to the text
> colour.{color}
> This works OK where the links form part of a menu system.
> In that case it's obvious that the the text items must be links.
> However it does not provide sufficient contrast where the links are part of
> body text.
> This is particularly true when a whole section of text is a link, so there
> is no contrast within the text.
> Users should not have to search to find the links.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7178) Is kafka compatible with zookeeper 3.5.x ?
[ https://issues.apache.org/jira/browse/KAFKA-7178?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7178: -- Component/s: zkclient > Is kafka compatible with zookeeper 3.5.x ? > -- > > Key: KAFKA-7178 > URL: https://issues.apache.org/jira/browse/KAFKA-7178 > Project: Kafka > Issue Type: Improvement > Components: zkclient >Reporter: fwq >Priority: Major > > Hi, all > I want to know is kafka versions (0.8.x, 0.9.x, 0.10.x 0.11.x 1.x) compatible > with zookeeper 3.5.x with dynamic reconfiguration feature? > some refs on here: > https://serverfault.com/questions/854650/kafka-compatible-with-zookeeper-3-5-feature-rebalancing-client-connections -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7309) Upgrade Jacoco for Java 11 support
[ https://issues.apache.org/jira/browse/KAFKA-7309?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7309: -- Component/s: packaging > Upgrade Jacoco for Java 11 support > -- > > Key: KAFKA-7309 > URL: https://issues.apache.org/jira/browse/KAFKA-7309 > Project: Kafka > Issue Type: Sub-task > Components: packaging >Reporter: Ismael Juma >Priority: Major > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7276) Consider using re2j to speed up regex operations
[ https://issues.apache.org/jira/browse/KAFKA-7276?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7276: -- Component/s: packaging > Consider using re2j to speed up regex operations > > > Key: KAFKA-7276 > URL: https://issues.apache.org/jira/browse/KAFKA-7276 > Project: Kafka > Issue Type: Task > Components: packaging >Reporter: Ted Yu >Priority: Major > > https://github.com/google/re2j > re2j claims to do linear time regular expression matching in Java. > Its benefit is most obvious for deeply nested regex (such as a | b | c | d). > We should consider using re2j to speed up regex operations. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-5891) Cast transformation fails if record schema contains timestamp field
[
https://issues.apache.org/jira/browse/KAFKA-5891?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586573#comment-16586573
]
ASF GitHub Bot commented on KAFKA-5891:
---
rayokota opened a new pull request #5537: KAFKA-5891: Adapts #4633 with schema
tests
URL: https://github.com/apache/kafka/pull/5537
This adapts PR #4633 from @maver1ck with suggested unit test from @rhauch
### 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 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
> Cast transformation fails if record schema contains timestamp field
> ---
>
> Key: KAFKA-5891
> URL: https://issues.apache.org/jira/browse/KAFKA-5891
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Affects Versions: 0.11.0.0
>Reporter: Artem Plotnikov
>Priority: Major
>
> I have the following simple type cast transformation:
> {code}
> name=postgresql-source-simple
> connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
> tasks.max=1
> connection.url=jdbc:postgresql://localhost:5432/testdb?user=postgres=mysecretpassword
> query=SELECT 1::INT as a, '2017-09-14 10:23:54'::TIMESTAMP as b
> transforms=Cast
> transforms.Cast.type=org.apache.kafka.connect.transforms.Cast$Value
> transforms.Cast.spec=a:boolean
> mode=bulk
> topic.prefix=clients
> {code}
> Which fails with the following exception in runtime:
> {code}
> [2017-09-14 16:51:01,885] ERROR Task postgresql-source-simple-0 threw an
> uncaught and unrecoverable exception
> (org.apache.kafka.connect.runtime.WorkerTask:148)
> org.apache.kafka.connect.errors.DataException: Invalid Java object for schema
> type INT64: class java.sql.Timestamp for field: "null"
> at
> org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:239)
> at
> org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:209)
> at org.apache.kafka.connect.data.Struct.put(Struct.java:214)
> at
> org.apache.kafka.connect.transforms.Cast.applyWithSchema(Cast.java:152)
> at org.apache.kafka.connect.transforms.Cast.apply(Cast.java:108)
> at
> org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38)
> at
> org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:190)
> at
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:168)
> at
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
> at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {code}
> If I remove the transforms.* part of the connector it will work correctly.
> Actually, it doesn't really matter which types I use in the transformation
> for field 'a', just the existence of a timestamp field brings the exception.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-6706) NETWORK_EXCEPTION and REQUEST_TIMED_OUT in mirrormaker producer after 1.0 broker upgrade
[
https://issues.apache.org/jira/browse/KAFKA-6706?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586534#comment-16586534
]
David van Geest commented on KAFKA-6706:
We also noticed this regression once we updated to 1.0.1 from 0.11.0.1. Not
MirrorMaker in our case, but a number of different apps using a variety of
clients (definitely the official Java client, but others as well).
We are currently testing 1.1.1, and it does not seem to have the same problems.
> NETWORK_EXCEPTION and REQUEST_TIMED_OUT in mirrormaker producer after 1.0
> broker upgrade
>
>
> Key: KAFKA-6706
> URL: https://issues.apache.org/jira/browse/KAFKA-6706
> Project: Kafka
> Issue Type: Bug
> Components: core, network
>Affects Versions: 1.0.0
>Reporter: Di Shang
>Priority: Blocker
> Labels: mirror-maker
>
> We have 2 clusters A and B with 4 brokers each, we use mirrormaker to
> replicate topics from A to B.
> We recently upgraded our brokers from 0.10.2.0 to 1.0.0, after the upgrade
> we started seeing the mirrormaker task showing producer errors and
> intermittently dying.
> We tried using 1.0.0 and 0.10.2.0 mirrormaker, both have the same problem.
> Downgrading cluster B brokers back to 0.10.2.0 and the problem went away, so
> we think it's a server side problem.
> There are 2 types of errors: REQUEST_TIMED_OUT and NETWORK_EXCEPTION. For
> testing, I used a topic *logging* with 20 partitions and 3 replicas (same on
> cluster A and B), the source topic has 50+ million msg.
> (this is from mirrormaker 1.0 at info level, the 0.10.2.0 log is very similar)
> {noformat}
> 22 Mar 2018 02:16:07.407 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 35122 on
> topic-partition logging-7, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:17:49.731 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 51572 on
> topic-partition logging-7, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:18:33.903 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 57785 on
> topic-partition logging-5, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:21:21.399 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 85406 on
> topic-partition logging-18, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:25:22.278 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 128047 on
> topic-partition logging-5, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:26:17.154 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 137049 on
> topic-partition logging-18, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:27:57.358 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 153976 on
> topic-partition logging-5, retrying (2147483646 attempts left). Error:
> REQUEST_TIMED_OUT
> 22 Mar 2018 02:27:57.779 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 154077 on
> topic-partition logging-2, retrying (2147483646 attempts left). Error:
> NETWORK_EXCEPTION
> 22 Mar 2018 02:27:57.780 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
> clientId=producer-1] Got error produce response with correlation id 154077 on
> topic-partition logging-10, retrying (2147483646 attempts left). Error:
> NETWORK_EXCEPTION
> 22 Mar 2018 02:27:57.780 [kafka-producer-network-thread | producer-1] WARN
> org.apache.kafka.clients.producer.internals.Sender warn(line:251) [Producer
[jira] [Commented] (KAFKA-6763) Consider using direct byte buffers in SslTransportLayer
[ https://issues.apache.org/jira/browse/KAFKA-6763?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586531#comment-16586531 ] Ismael Juma commented on KAFKA-6763: Yes, that's right [~leogomes]. One option is to simply run the existing producer/consumer perf tools after making the changes. > Consider using direct byte buffers in SslTransportLayer > --- > > Key: KAFKA-6763 > URL: https://issues.apache.org/jira/browse/KAFKA-6763 > Project: Kafka > Issue Type: Improvement > Components: network >Reporter: Ismael Juma >Priority: Minor > Labels: performance, tls > > We use heap byte buffers in SslTransportLayer. For netReadBuffer and > netWriteBuffer, it means that the NIO layer has to copy to/from a native > buffer before it can write/read to the socket. It would be good to test if > switching to direct byte buffers improves performance. We can't be sure as > the benefit of avoiding the copy could be offset by the specifics of the > operations we perform on netReadBuffer, netWriteBuffer and appReadBuffer. > We should benchmark produce and consume performance and try a few > combinations of direct/heap byte buffers for netReadBuffer, netWriteBuffer > and appReadBuffer (the latter should probably remain as a heap byte buffer, > but no harm in testing it too). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7314) MirrorMaker example in documentation does not work
[
https://issues.apache.org/jira/browse/KAFKA-7314?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-7314:
--
Component/s: mirrormaker
> MirrorMaker example in documentation does not work
> --
>
> Key: KAFKA-7314
> URL: https://issues.apache.org/jira/browse/KAFKA-7314
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Reporter: John Wilkinson
>Priority: Critical
>
> Kafka MirrorMaker as described in the documentation
> [here|https://kafka.apache.org/documentation/#basic_ops_mirror_maker] does
> not work. Instead of pulling messages from the consumer-defined
> {{bootstrap.servers}} and pushing to the producer-defined
> {{bootstrap.servers}}, it consumes and producers on the same topic on the
> same host repeatedly.
> To replicate, set up two instances of kafka following
> [this|https://docs.confluent.io/current/installation/docker/docs/installation/recipes/single-node-client.html]
> guide. The schema registry and rest proxy are unnecessary.
> [Here|https://hub.docker.com/r/confluentinc/cp-kafka/] is the DockerHub page
> for the image. The Kafka version is 2.0.0.
> Using those two kafka instances, go {{docker exec}} into one and set up the
> {{consumer.properties}} and the {{producer.properties}} following the
> MirrorMaker guide.
> Oddly, if you put in garbage unresolvable server addresses in the config,
> there will be an error, despite the configs not getting used.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7282) Failed to read `log header` from file channel
[
https://issues.apache.org/jira/browse/KAFKA-7282?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ray Chiang updated KAFKA-7282:
--
Component/s: log
> Failed to read `log header` from file channel
> -
>
> Key: KAFKA-7282
> URL: https://issues.apache.org/jira/browse/KAFKA-7282
> Project: Kafka
> Issue Type: Bug
> Components: log
>Affects Versions: 0.11.0.2, 1.1.1, 2.0.0
> Environment: Linux
>Reporter: Alastair Munro
>Priority: Major
>
> Full stack trace:
> {code:java}
> [2018-08-13 11:22:01,635] ERROR [ReplicaManager broker=2] Error processing
> fetch operation on partition segmenter-evt-v1-14, offset 96745
> (kafka.server.ReplicaManager)
> org.apache.kafka.common.KafkaException: java.io.EOFException: Failed to read
> `log header` from file channel `sun.nio.ch.FileChannelImpl@6e6d8ddd`.
> Expected to read 17 bytes, but reached end of file after reading 0 bytes.
> Started read from position 25935.
> at
> org.apache.kafka.common.record.RecordBatchIterator.makeNext(RecordBatchIterator.java:40)
> at
> org.apache.kafka.common.record.RecordBatchIterator.makeNext(RecordBatchIterator.java:24)
> at
> org.apache.kafka.common.utils.AbstractIterator.maybeComputeNext(AbstractIterator.java:79)
> at
> org.apache.kafka.common.utils.AbstractIterator.hasNext(AbstractIterator.java:45)
> at
> org.apache.kafka.common.record.FileRecords.searchForOffsetWithSize(FileRecords.java:286)
> at kafka.log.LogSegment.translateOffset(LogSegment.scala:254)
> at kafka.log.LogSegment.read(LogSegment.scala:277)
> 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:1837)
> 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:678)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:107)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)
> at java.lang.Thread.run(Thread.java:748)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-6763) Consider using direct byte buffers in SslTransportLayer
[ https://issues.apache.org/jira/browse/KAFKA-6763?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-6763: -- Component/s: network > Consider using direct byte buffers in SslTransportLayer > --- > > Key: KAFKA-6763 > URL: https://issues.apache.org/jira/browse/KAFKA-6763 > Project: Kafka > Issue Type: Improvement > Components: network >Reporter: Ismael Juma >Priority: Minor > Labels: performance, tls > > We use heap byte buffers in SslTransportLayer. For netReadBuffer and > netWriteBuffer, it means that the NIO layer has to copy to/from a native > buffer before it can write/read to the socket. It would be good to test if > switching to direct byte buffers improves performance. We can't be sure as > the benefit of avoiding the copy could be offset by the specifics of the > operations we perform on netReadBuffer, netWriteBuffer and appReadBuffer. > We should benchmark produce and consume performance and try a few > combinations of direct/heap byte buffers for netReadBuffer, netWriteBuffer > and appReadBuffer (the latter should probably remain as a heap byte buffer, > but no harm in testing it too). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7310) Fix SSL tests when running with Java 11
[ https://issues.apache.org/jira/browse/KAFKA-7310?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7310: -- Component/s: unit tests packaging > Fix SSL tests when running with Java 11 > --- > > Key: KAFKA-7310 > URL: https://issues.apache.org/jira/browse/KAFKA-7310 > Project: Kafka > Issue Type: Sub-task > Components: packaging, unit tests >Reporter: Ismael Juma >Priority: Critical > Fix For: 2.1.0 > > > * Many SSL tests in clients fail. Probably related to the TLS 1.3 changes in > Java 11. > * Many SSL (and some SASL) tests in Core fail. Maybe same underlying reason > as the clients failures, but I didn't investigate. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7232) Allow kafka-topics.sh to take brokerid as parameter to show partitions associated with it
[ https://issues.apache.org/jira/browse/KAFKA-7232?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ray Chiang updated KAFKA-7232: -- Component/s: tools > Allow kafka-topics.sh to take brokerid as parameter to show partitions > associated with it > - > > Key: KAFKA-7232 > URL: https://issues.apache.org/jira/browse/KAFKA-7232 > Project: Kafka > Issue Type: Improvement > Components: tools >Reporter: Ratish Ravindran >Priority: Minor > > Currently with kafka-topics.sh if we want to get the list of partitions > associated with a specific broker irrespective of whether it is leader or > replica, we pipe the output and then do grep on it. > I am proposing the change to add option in TopicCommand.scala to pass the > broker id. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
[ https://issues.apache.org/jira/browse/KAFKA-7315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586476#comment-16586476 ] Jordan Moore commented on KAFKA-7315: - Thanks for opening [~vvcephei] Thoughts: List out the options in the documentation, don't force a developer towards using a Schema Registry (since this is AK docs, not Confluent or Hortonworks for example). So, options: # Take an Avro object and push it as raw bytes directly; [example here|http://aseigneurin.github.io/2016/03/04/kafka-spark-avro-producing-and-consuming-avro-messages.html] # Use a Schema Registry, and document how it works, in general (using referential lookups to external sources within the SerDe interfaces) ## Possibly link to (or simply mention) known implementations --- [Confluent Schema Registry|https://www.confluent.io/confluent-schema-registry/] --- [Hortonworks Registry|https://registry-project.readthedocs.io/en/latest/] --- [Cloudera|http://blog.cloudera.com/blog/2018/07/robust-message-serialization-in-apache-kafka-using-apache-avro-part-2/] > Streams serialization docs contain a broken link for Avro > - > > Key: KAFKA-7315 > URL: https://issues.apache.org/jira/browse/KAFKA-7315 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > Labels: docuentation, newbie > > https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-6312) Add documentation about kafka-consumer-groups.sh's ability to set/change offsets
[
https://issues.apache.org/jira/browse/KAFKA-6312?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586472#comment-16586472
]
ASF GitHub Bot commented on KAFKA-6312:
---
hachikuji closed pull request #4527: KAFKA-6312 Add documentation about
kafka-consumer-groups.sh's ability…
URL: https://github.com/apache/kafka/pull/4527
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/ops.html b/docs/ops.html
index 95b9a960172..6fe8fffeba2 100644
--- a/docs/ops.html
+++ b/docs/ops.html
@@ -229,6 +229,69 @@ Managing C
Deletion of requested consumer groups ('my-group', 'my-other-group') was
successful.
+
+ To reset offsets of a consumer group, "--reset-offsets" option can be used.
+ This option supports one consumer group at the time. It requires defining
following scopes: --all-topics or --topic. One scope must be selected, unless
you use '--from-file' scenario. Also, first make sure that the consumer
instances are inactive.
+ See https://cwiki.apache.org/confluence/display/KAFKA/KIP-122%3A+Add+Reset+Consumer+Group+Offsets+tooling;>KIP-122
for more details.
+
+
+ It has 3 execution options:
+
+
+ (default) to display which offsets to reset.
+
+
+ --execute : to execute --reset-offsets process.
+
+
+ --export : to export the results to a CSV format.
+
+
+
+
+ --reset-offsets also has following scenarios to choose from (atleast one
scenario must be selected):
+
+
+ --to-datetime String: datetime : Reset offsets to offsets from
datetime. Format: '-MM-DDTHH:mm:SS.sss'
+
+
+ --to-earliest : Reset offsets to earliest offset.
+
+
+ --to-latest : Reset offsets to latest offset.
+
+
+ --shift-by Long: number-of-offsets : Reset offsets shifting
current offset by 'n', where 'n' can be positive or negative.
+
+
+ --from-file : Reset offsets to values defined in CSV file.
+
+
+ --to-current : Resets offsets to current offset.
+
+
+ --by-duration String: duration : Reset offsets to offset by
duration from current timestamp. Format: 'PnDTnHnMnS'
+
+
+ --to-offset : Reset offsets to a specific offset.
+
+
+
+ Please note, that out of range offsets will be adjusted to available offset
end. For example, if offset end is at 10 and offset shift request is
+ of 15, then, offset at 10 will actually be selected.
+
+
+ For example, to reset offsets of a consumer group to the latest offset:
+
+
+ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092
--reset-offsets --group consumergroup1 --topic topic1 --to-latest
+
+ TOPIC PARTITION NEW-OFFSET
+ topic1 0 0
+
+
+
+
If you are using the old high-level consumer and storing the group metadata
in ZooKeeper (i.e. offsets.storage=zookeeper), pass
--zookeeper instead of bootstrap-server:
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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
> Add documentation about kafka-consumer-groups.sh's ability to set/change
> offsets
>
>
> Key: KAFKA-6312
> URL: https://issues.apache.org/jira/browse/KAFKA-6312
> Project: Kafka
> Issue Type: Improvement
> Components: documentation
>Reporter: James Cheng
>Assignee: Mayank Tankhiwale
>Priority: Major
> Labels: newbie
>
> KIP-122 added the ability for kafka-consumer-groups.sh to reset/change
> consumer offsets, at a fine grained level.
> There is documentation on it in the kafka-consumer-groups.sh usage text.
> There is no such documentation on the kafka.apache.org website. We should add
> some documentation to the website, so that users can read about the
> functionality without having the tools installed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-7315) Streams serialization docs contain a broken link for Avro
John Roesler created KAFKA-7315: --- Summary: Streams serialization docs contain a broken link for Avro Key: KAFKA-7315 URL: https://issues.apache.org/jira/browse/KAFKA-7315 Project: Kafka Issue Type: Improvement Components: streams Reporter: John Roesler https://kafka.apache.org/documentation/streams/developer-guide/datatypes.html#avro -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7304) memory leakage in org.apache.kafka.common.network.Selector
[
https://issues.apache.org/jira/browse/KAFKA-7304?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yu Yang updated KAFKA-7304:
---
Affects Version/s: 1.1.1
> memory leakage in org.apache.kafka.common.network.Selector
> --
>
> Key: KAFKA-7304
> URL: https://issues.apache.org/jira/browse/KAFKA-7304
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 1.1.0, 1.1.1
>Reporter: Yu Yang
>Priority: Critical
> Fix For: 1.1.2, 2.0.1, 2.1.0
>
> Attachments: 7304.v4.txt, Screen Shot 2018-08-16 at 11.04.16 PM.png,
> Screen Shot 2018-08-16 at 11.06.38 PM.png, Screen Shot 2018-08-16 at 12.41.26
> PM.png, Screen Shot 2018-08-16 at 4.26.19 PM.png, Screen Shot 2018-08-17 at
> 1.03.35 AM.png, Screen Shot 2018-08-17 at 1.04.32 AM.png, Screen Shot
> 2018-08-17 at 1.05.30 AM.png
>
>
> We are testing secured writing to kafka through ssl. Testing at small scale,
> ssl writing to kafka was fine. However, when we enabled ssl writing at a
> larger scale (>40k clients write concurrently), the kafka brokers soon hit
> OutOfMemory issue with 4G memory setting. We have tried with increasing the
> heap size to 10Gb, but encountered the same issue.
> We took a few heap dumps , and found that most of the heap memory is
> referenced through org.apache.kafka.common.network.Selector objects. There
> are two Channel maps field in Selector. It seems that somehow the objects is
> not deleted from the map in a timely manner.
> One observation is that the memory leak seems relate to kafka partition
> leader changes. If there is broker restart etc. in the cluster that caused
> partition leadership change, the brokers may hit the OOM issue faster.
> {code}
> private final Map channels;
> private final Map closingChannels;
> {code}
> Please see the attached images and the following link for sample gc
> analysis.
> http://gceasy.io/my-gc-report.jsp?p=c2hhcmVkLzIwMTgvMDgvMTcvLS1nYy5sb2cuMC5jdXJyZW50Lmd6LS0yLTM5LTM0
> the command line for running kafka:
> {code}
> java -Xms10g -Xmx10g -XX:NewSize=512m -XX:MaxNewSize=512m
> -Xbootclasspath/p:/usr/local/libs/bcp -XX:MetaspaceSize=128m -XX:+UseG1GC
> -XX:MaxGCPauseMillis=25 -XX:InitiatingHeapOccupancyPercent=35
> -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=25
> -XX:MaxMetaspaceFreeRatio=75 -XX:+PrintGCDetails -XX:+PrintGCDateStamps
> -XX:+PrintTenuringDistribution -Xloggc:/var/log/kafka/gc.log
> -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=40 -XX:GCLogFileSize=50M
> -Djava.awt.headless=true
> -Dlog4j.configuration=file:/etc/kafka/log4j.properties
> -Dcom.sun.management.jmxremote
> -Dcom.sun.management.jmxremote.authenticate=false
> -Dcom.sun.management.jmxremote.ssl=false
> -Dcom.sun.management.jmxremote.port=
> -Dcom.sun.management.jmxremote.rmi.port= -cp /usr/local/libs/*
> kafka.Kafka /etc/kafka/server.properties
> {code}
> We use java 1.8.0_102, and has applied a TLS patch on reducing
> X509Factory.certCache map size from 750 to 20.
> {code}
> java -version
> java version "1.8.0_102"
> Java(TM) SE Runtime Environment (build 1.8.0_102-b14)
> Java HotSpot(TM) 64-Bit Server VM (build 25.102-b14, mixed mode)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-7311) Sender should reset next batch expiry time between poll loops
[ https://issues.apache.org/jira/browse/KAFKA-7311?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Guozhang Wang resolved KAFKA-7311. -- Resolution: Fixed > Sender should reset next batch expiry time between poll loops > - > > Key: KAFKA-7311 > URL: https://issues.apache.org/jira/browse/KAFKA-7311 > Project: Kafka > Issue Type: Bug > Components: clients >Affects Versions: 2.1.0 >Reporter: Rohan Desai >Assignee: Rohan Desai >Priority: Major > Fix For: 2.1.0 > > > Sender/RecordAccumulator never resets the next batch expiry time. Its always > computed as the min of the current value and the expiry time for all batches > being processed. This means that its always set to the expiry time of the > first batch, and once that time has passed Sender starts spinning on epoll > with a timeout of 0, which consumes a lot of CPU. This patch updates Sender > to reset the next batch expiry time on each poll loop so that a new value > reflecting the expiry time for the current set of batches is computed. We > observed this running KSQL when investigating why throughput would drop after > about 10 minutes (the default delivery timeout). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7314) MirrorMaker example in documentation does not work
[
https://issues.apache.org/jira/browse/KAFKA-7314?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
John Wilkinson updated KAFKA-7314:
--
Description:
Kafka MirrorMaker as described in the documentation
[here|https://kafka.apache.org/documentation/#basic_ops_mirror_maker] does not
work. Instead of pulling messages from the consumer-defined
{{bootstrap.servers}} and pushing to the producer-defined
{{bootstrap.servers}}, it consumes and producers on the same topic on the same
host repeatedly.
To replicate, set up two instances of kafka following
[this|https://docs.confluent.io/current/installation/docker/docs/installation/recipes/single-node-client.html]
guide. The schema registry and rest proxy are unnecessary.
[Here|https://hub.docker.com/r/confluentinc/cp-kafka/] is the DockerHub page
for the image. The Kafka version is 2.0.0.
Using those two kafka instances, go {{docker exec}} into one and set up the
{{consumer.properties}} and the {{producer.properties}} following the
MirrorMaker guide.
Oddly, if you put in garbage unresolvable server addresses in the config, there
will be an error, despite the configs not getting used.
was:
Kafka MirrorMaker as described in the documentation
[here|https://kafka.apache.org/documentation/#basic_ops_mirror_maker] does not
work. Instead of pulling messages from the consumer-defined
{{bootstrap.servers}} and pushing to the producer-defined
{{bootstrap.servers}}, it consumes and producers on the same topic on the same
host repeatedly.
To replicate, set up two instances of kafka following
[this|https://docs.confluent.io/current/installation/docker/docs/installation/recipes/single-node-client.html]
guide. The schema registry and rest proxy are unnecessary.
[Here|https://hub.docker.com/r/confluentinc/cp-kafka/] is the DockerHub page
for the image. The Kafka version is 2.1.0.
Using those two kafka instances, go {{docker exec}} into one and set up the
{{consumer.properties}} and the {{producer.properties}} following the
MirrorMaker guide.
Oddly, if you put in garbage unresolvable server addresses in the config, there
will be an error, despite the configs not getting used.
> MirrorMaker example in documentation does not work
> --
>
> Key: KAFKA-7314
> URL: https://issues.apache.org/jira/browse/KAFKA-7314
> Project: Kafka
> Issue Type: Bug
>Reporter: John Wilkinson
>Priority: Critical
>
> Kafka MirrorMaker as described in the documentation
> [here|https://kafka.apache.org/documentation/#basic_ops_mirror_maker] does
> not work. Instead of pulling messages from the consumer-defined
> {{bootstrap.servers}} and pushing to the producer-defined
> {{bootstrap.servers}}, it consumes and producers on the same topic on the
> same host repeatedly.
> To replicate, set up two instances of kafka following
> [this|https://docs.confluent.io/current/installation/docker/docs/installation/recipes/single-node-client.html]
> guide. The schema registry and rest proxy are unnecessary.
> [Here|https://hub.docker.com/r/confluentinc/cp-kafka/] is the DockerHub page
> for the image. The Kafka version is 2.0.0.
> Using those two kafka instances, go {{docker exec}} into one and set up the
> {{consumer.properties}} and the {{producer.properties}} following the
> MirrorMaker guide.
> Oddly, if you put in garbage unresolvable server addresses in the config,
> there will be an error, despite the configs not getting used.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7278) replaceSegments() should not call asyncDeleteSegment() for segments which have been removed from segments list
[ https://issues.apache.org/jira/browse/KAFKA-7278?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dong Lin updated KAFKA-7278: Fix Version/s: 2.1.0 1.1.2 > replaceSegments() should not call asyncDeleteSegment() for segments which > have been removed from segments list > -- > > Key: KAFKA-7278 > URL: https://issues.apache.org/jira/browse/KAFKA-7278 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > Fix For: 1.1.2, 2.0.1, 2.1.0 > > > Currently Log.replaceSegments() will call `asyncDeleteSegment(...)` for every > segment listed in the `oldSegments`. oldSegments should be constructed from > Log.segments and only contain segments listed in Log.segments. > However, Log.segments may be modified between the time oldSegments is > determined to the time Log.replaceSegments() is called. If there are > concurrent async deletion of the same log segment file, Log.replaceSegments() > will call asyncDeleteSegment() for a segment that does not exist and Kafka > server may shutdown the log directory due to NoSuchFileException. > This is likely the root cause of > https://issues.apache.org/jira/browse/KAFKA-6188. > Given the understanding of the problem, we should be able to fix the issue by > only deleting segment if the segment can be found in Log.segments. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7314) MirrorMaker example in documentation does not work
John Wilkinson created KAFKA-7314:
-
Summary: MirrorMaker example in documentation does not work
Key: KAFKA-7314
URL: https://issues.apache.org/jira/browse/KAFKA-7314
Project: Kafka
Issue Type: Bug
Reporter: John Wilkinson
Kafka MirrorMaker as described in the documentation
[here|https://kafka.apache.org/documentation/#basic_ops_mirror_maker] does not
work. Instead of pulling messages from the consumer-defined
{{bootstrap.servers}} and pushing to the producer-defined
{{bootstrap.servers}}, it consumes and producers on the same topic on the same
host repeatedly.
To replicate, set up two instances of kafka following
[this|https://docs.confluent.io/current/installation/docker/docs/installation/recipes/single-node-client.html]
guide. The schema registry and rest proxy are unnecessary.
[Here|https://hub.docker.com/r/confluentinc/cp-kafka/] is the DockerHub page
for the image. The Kafka version is 2.1.0.
Using those two kafka instances, go {{docker exec}} into one and set up the
{{consumer.properties}} and the {{producer.properties}} following the
MirrorMaker guide.
Oddly, if you put in garbage unresolvable server addresses in the config, there
will be an error, despite the configs not getting used.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-7278) replaceSegments() should not call asyncDeleteSegment() for segments which have been removed from segments list
[ https://issues.apache.org/jira/browse/KAFKA-7278?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586362#comment-16586362 ] ASF GitHub Bot commented on KAFKA-7278: --- lindong28 opened a new pull request #5535: Cherry-pick KAFKA-7278; replaceSegments() should not call asyncDeleteSegment() for segments which have been removed from segments list URL: https://github.com/apache/kafka/pull/5535 ### 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 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 > replaceSegments() should not call asyncDeleteSegment() for segments which > have been removed from segments list > -- > > Key: KAFKA-7278 > URL: https://issues.apache.org/jira/browse/KAFKA-7278 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > Fix For: 2.0.1 > > > Currently Log.replaceSegments() will call `asyncDeleteSegment(...)` for every > segment listed in the `oldSegments`. oldSegments should be constructed from > Log.segments and only contain segments listed in Log.segments. > However, Log.segments may be modified between the time oldSegments is > determined to the time Log.replaceSegments() is called. If there are > concurrent async deletion of the same log segment file, Log.replaceSegments() > will call asyncDeleteSegment() for a segment that does not exist and Kafka > server may shutdown the log directory due to NoSuchFileException. > This is likely the root cause of > https://issues.apache.org/jira/browse/KAFKA-6188. > Given the understanding of the problem, we should be able to fix the issue by > only deleting segment if the segment can be found in Log.segments. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7190) Under low traffic conditions purging repartition topics cause WARN statements about UNKNOWN_PRODUCER_ID
[ https://issues.apache.org/jira/browse/KAFKA-7190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586359#comment-16586359 ] Jason Gustafson commented on KAFKA-7190: Posted a KIP here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-360%3A+Improve+handling+of+unknown+producer. Please take a look. > Under low traffic conditions purging repartition topics cause WARN statements > about UNKNOWN_PRODUCER_ID > - > > Key: KAFKA-7190 > URL: https://issues.apache.org/jira/browse/KAFKA-7190 > Project: Kafka > Issue Type: Improvement > Components: core, streams >Affects Versions: 1.1.0, 1.1.1 >Reporter: Bill Bejeck >Assignee: lambdaliu >Priority: Major > > When a streams application has little traffic, then it is possible that > consumer purging would delete > even the last message sent by a producer (i.e., all the messages sent by > this producer have been consumed and committed), and as a result, the broker > would delete that producer's ID. The next time when this producer tries to > send, it will get this UNKNOWN_PRODUCER_ID error code, but in this case, > this error is retriable: the producer would just get a new producer id and > retries, and then this time it will succeed. > > Possible fixes could be on the broker side, i.e., delaying the deletion of > the produderIDs for a more extended period or on the streams side developing > a more conservative approach to deleting offsets from repartition topics > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7298) Concurrent DeleteRecords can lead to fatal OutOfSequence error in producer
[ https://issues.apache.org/jira/browse/KAFKA-7298?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586339#comment-16586339 ] Matthias J. Sax commented on KAFKA-7298: [~hachikuji] Should this be back ported to older branches, too? > Concurrent DeleteRecords can lead to fatal OutOfSequence error in producer > -- > > Key: KAFKA-7298 > URL: https://issues.apache.org/jira/browse/KAFKA-7298 > Project: Kafka > Issue Type: Bug >Reporter: Jason Gustafson >Assignee: Jason Gustafson >Priority: Major > Fix For: 2.0.1 > > > We have logic in the producer to handle unknown producer errors. Basically > when the producer gets an unknown producer error, it checks whether the log > start offset is larger than the last acknowledged offset. If it is, then we > know the error is spurious and we reset the sequence number to 0, which the > broker will then accept. > It can happen after a DeleteRecords call, however, that the only record > remaining in the log is a transaction marker, which does not have a sequence > number. The error we get in this case is OUT_OF_SEQUENCE rather than > UNKNOWN_PRODUCER, which is fatal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-7278) replaceSegments() should not call asyncDeleteSegment() for segments which have been removed from segments list
[ https://issues.apache.org/jira/browse/KAFKA-7278?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jason Gustafson resolved KAFKA-7278. Resolution: Fixed Fix Version/s: 2.0.1 > replaceSegments() should not call asyncDeleteSegment() for segments which > have been removed from segments list > -- > > Key: KAFKA-7278 > URL: https://issues.apache.org/jira/browse/KAFKA-7278 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > Fix For: 2.0.1 > > > Currently Log.replaceSegments() will call `asyncDeleteSegment(...)` for every > segment listed in the `oldSegments`. oldSegments should be constructed from > Log.segments and only contain segments listed in Log.segments. > However, Log.segments may be modified between the time oldSegments is > determined to the time Log.replaceSegments() is called. If there are > concurrent async deletion of the same log segment file, Log.replaceSegments() > will call asyncDeleteSegment() for a segment that does not exist and Kafka > server may shutdown the log directory due to NoSuchFileException. > This is likely the root cause of > https://issues.apache.org/jira/browse/KAFKA-6188. > Given the understanding of the problem, we should be able to fix the issue by > only deleting segment if the segment can be found in Log.segments. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7278) replaceSegments() should not call asyncDeleteSegment() for segments which have been removed from segments list
[
https://issues.apache.org/jira/browse/KAFKA-7278?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586320#comment-16586320
]
ASF GitHub Bot commented on KAFKA-7278:
---
hachikuji closed pull request #5491: KAFKA-7278; replaceSegments() should not
call asyncDeleteSegment() for segments which have been removed from segments
list
URL: https://github.com/apache/kafka/pull/5491
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/scala/kafka/log/Log.scala
b/core/src/main/scala/kafka/log/Log.scala
index e4be8fcc43d..afe151d69b6 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -1740,7 +1740,9 @@ class Log(@volatile var dir: File,
}
/**
- * Perform an asynchronous delete on the given file if it exists (otherwise
do nothing)
+ * Perform an asynchronous delete on the given file.
+ *
+ * This method assumes that the file exists and the method is not
thread-safe.
*
* This method does not need to convert IOException (thrown from
changeFileSuffixes) to KafkaStorageException because
* it is either called before all logs are loaded or the caller will catch
and handle IOException
@@ -1791,10 +1793,13 @@ class Log(@volatile var dir: File,
* @param isRecoveredSwapFile true if the new segment was created from a
swap file during recovery after a crash
*/
private[log] def replaceSegments(newSegments: Seq[LogSegment], oldSegments:
Seq[LogSegment], isRecoveredSwapFile: Boolean = false) {
-val sortedNewSegments = newSegments.sortBy(_.baseOffset)
-val sortedOldSegments = oldSegments.sortBy(_.baseOffset)
-
lock synchronized {
+ val sortedNewSegments = newSegments.sortBy(_.baseOffset)
+ // Some old segments may have been removed from index and scheduled for
async deletion after the caller reads segments
+ // but before this method is executed. We want to filter out those
segments to avoid calling asyncDeleteSegment()
+ // multiple times for the same segment.
+ val sortedOldSegments = oldSegments.filter(seg =>
segments.containsKey(seg.baseOffset)).sortBy(_.baseOffset)
+
checkIfMemoryMappedBufferClosed()
// need to do this in two phases to be crash safe AND do the delete
asynchronously
// if we crash in the middle of this we complete the swap in
loadSegments()
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index b351311b329..0240707ca3b 100755
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -21,6 +21,7 @@ import java.io.{File, RandomAccessFile}
import java.nio._
import java.nio.file.Paths
import java.util.Properties
+import java.util.concurrent.{CountDownLatch, TimeUnit}
import kafka.common._
import kafka.server.{BrokerTopicStats, LogDirFailureChannel}
@@ -89,6 +90,74 @@ class LogCleanerTest extends JUnitSuite {
assertEquals(expectedBytesRead, stats.bytesRead)
}
+ @Test
+ def testCleanSegmentsWithConcurrentSegmentDeletion(): Unit = {
+val deleteStartLatch = new CountDownLatch(1)
+val deleteCompleteLatch = new CountDownLatch(1)
+
+// Construct a log instance. The replaceSegments() method of the log
instance is overridden so that
+// it waits for another thread to execute deleteOldSegments()
+val logProps = new Properties()
+logProps.put(LogConfig.SegmentBytesProp, 1024 : java.lang.Integer)
+logProps.put(LogConfig.CleanupPolicyProp, LogConfig.Compact + "," +
LogConfig.Delete)
+val topicPartition = Log.parseTopicPartitionName(dir)
+val producerStateManager = new ProducerStateManager(topicPartition, dir)
+val log = new Log(dir,
+ config = LogConfig.fromProps(logConfig.originals,
logProps),
+ logStartOffset = 0L,
+ recoveryPoint = 0L,
+ scheduler = time.scheduler,
+ brokerTopicStats = new BrokerTopicStats, time,
+ maxProducerIdExpirationMs = 60 * 60 * 1000,
+ producerIdExpirationCheckIntervalMs =
LogManager.ProducerIdExpirationCheckIntervalMs,
+ topicPartition = topicPartition,
+ producerStateManager = producerStateManager,
+ logDirFailureChannel = new LogDirFailureChannel(10)) {
+ override def replaceSegments(newSegments: Seq[LogSegment], oldSegments:
Seq[LogSegment], isRecoveredSwapFile: Boolean = false): Unit = {
+deleteStartLatch.countDown()
+if (!deleteCompleteLatch.await(5000, TimeUnit.MILLISECONDS)) {
+
[jira] [Commented] (KAFKA-7303) Kafka client is stuck when specifying wrong brokers
[
https://issues.apache.org/jira/browse/KAFKA-7303?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586255#comment-16586255
]
Chun Zhang commented on KAFKA-7303:
---
Thanks Manikumar. I will give it a try with 2.0
This bug is duplicated with many other issues inducing KAFKA-1894 and is fixed
by KIP-266.
> Kafka client is stuck when specifying wrong brokers
> ---
>
> Key: KAFKA-7303
> URL: https://issues.apache.org/jira/browse/KAFKA-7303
> Project: Kafka
> Issue Type: Bug
> Components: consumer
>Affects Versions: 1.1.0
>Reporter: Chun Zhang
>Priority: Major
>
> {code:java}
> import java.util.Collections;
> import java.util.Properties;
> import org.apache.kafka.clients.CommonClientConfigs;
> import org.apache.kafka.clients.consumer.ConsumerConfig;
> import org.apache.kafka.clients.consumer.ConsumerRecord;
> import org.apache.kafka.clients.consumer.ConsumerRecords;
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> public class KafkaBug {
> public static void main(String[] args) throws Exception {
> Properties props = new Properties();
> // intentionally use an irrelevant address
> props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
> "issues.apache.org:80");
> props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
> props.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id_string");
> props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.subscribe(Collections.singleton("mytopic"));
> // This call will block forever.
> consumer.poll(1000);
> }
> }
> {code}
> When I run the code above, I keep getting the error log below:
> {code:java}
> DEBUG [main] (21:21:25,959) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Connection with issues.apache.org/207.244.88.139
> disconnected
> java.net.ConnectException: Connection timed out
> at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
> at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
> at
> org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
> at
> org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:106)
> at
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:470)
> at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:261)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:156)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:228)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:205)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:279)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1149)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1115)
> at com.twosigma.example.kafka.bug.KafkaBug.main(KafkaBug.java:46)
> DEBUG [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Node -1 disconnected.
> WARN [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Connection to node -1 could not be established.
> Broker may not be available.
> DEBUG [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Give up sending metadata request since no node is
> available
> DEBUG [main] (21:21:26,013) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Give up sending metadata request since no node is
> available
> {code}
> I expect the program to fail when the wrong broker is specified.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-7303) Kafka client is stuck when specifying wrong brokers
[
https://issues.apache.org/jira/browse/KAFKA-7303?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chun Zhang resolved KAFKA-7303.
---
Resolution: Duplicate
> Kafka client is stuck when specifying wrong brokers
> ---
>
> Key: KAFKA-7303
> URL: https://issues.apache.org/jira/browse/KAFKA-7303
> Project: Kafka
> Issue Type: Bug
> Components: consumer
>Affects Versions: 1.1.0
>Reporter: Chun Zhang
>Priority: Major
>
> {code:java}
> import java.util.Collections;
> import java.util.Properties;
> import org.apache.kafka.clients.CommonClientConfigs;
> import org.apache.kafka.clients.consumer.ConsumerConfig;
> import org.apache.kafka.clients.consumer.ConsumerRecord;
> import org.apache.kafka.clients.consumer.ConsumerRecords;
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> public class KafkaBug {
> public static void main(String[] args) throws Exception {
> Properties props = new Properties();
> // intentionally use an irrelevant address
> props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
> "issues.apache.org:80");
> props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
> props.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id_string");
> props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.subscribe(Collections.singleton("mytopic"));
> // This call will block forever.
> consumer.poll(1000);
> }
> }
> {code}
> When I run the code above, I keep getting the error log below:
> {code:java}
> DEBUG [main] (21:21:25,959) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Connection with issues.apache.org/207.244.88.139
> disconnected
> java.net.ConnectException: Connection timed out
> at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
> at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
> at
> org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50)
> at
> org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:106)
> at
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:470)
> at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:261)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:156)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:228)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:205)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:279)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1149)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1115)
> at com.twosigma.example.kafka.bug.KafkaBug.main(KafkaBug.java:46)
> DEBUG [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Node -1 disconnected.
> WARN [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Connection to node -1 could not be established.
> Broker may not be available.
> DEBUG [main] (21:21:25,963) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Give up sending metadata request since no node is
> available
> DEBUG [main] (21:21:26,013) - [Consumer clientId=consumer-1,
> groupId=group_id_string] Give up sending metadata request since no node is
> available
> {code}
> I expect the program to fail when the wrong broker is specified.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Updated] (KAFKA-7313) KAFKA-7313; StopReplicaRequest should attempt to remove future replica for the partition only if future replica exists
[ https://issues.apache.org/jira/browse/KAFKA-7313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dong Lin updated KAFKA-7313: Description: This patch fixes two issues: 1) Currently if a broker received StopReplicaRequest with delete=true for the same offline replica, the first StopRelicaRequest will show KafkaStorageException and the second StopRelicaRequest will show ReplicaNotAvailableException. This is because the first StopRelicaRequest will remove the mapping (tp -> ReplicaManager.OfflinePartition) from ReplicaManager.allPartitions before returning KafkaStorageException, thus the second StopRelicaRequest will not find this partition as offline. This result appears to be inconsistent. And since the replica is already offline and broker will not be able to delete file for this replica, the StopReplicaRequest should fail without making any change and broker should still remember that this replica is offline. 2) Currently if broker receives StopReplicaRequest with delete=true, the broker will attempt to remove future replica for the partition, which will cause KafkaStorageException in the StopReplicaResponse if this replica does not have future replica. It is problematic to always return KafkaStorageException in the response if future replica does not exist. was: This patch fixes two issues: 1) Currently if a broker received StopReplicaRequest with delete=true for the same offline replica, the first StopRelicaRequest will show KafkaStorageException and the second StopRelicaRequest will show ReplicaNotAvailableException. This is because the first StopRelicaRequest will remove the mapping (tp -> ReplicaManager.OfflinePartition) from ReplicaManager.allPartitions before returning KafkaStorageException, thus the second StopRelicaRequest will not find this partition as offline. This result appears to be inconsistent. And since the replica is already offline and broker will not be able to delete file for this replica, the StopReplicaRequest should fail without making any change and broker should still remember that this replica is offline. 2) Currently if broker receives StopReplicaRequest with delete=true, the broker will attempt to remove future replica for the partition, which will cause KafkaStorageException in the StopReplicaResponse if this replica does not have future replica. It is problematic to always return KafkaStorageException in the response if future replica does not exist. > KAFKA-7313; StopReplicaRequest should attempt to remove future replica for > the partition only if future replica exists > -- > > Key: KAFKA-7313 > URL: https://issues.apache.org/jira/browse/KAFKA-7313 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > > This patch fixes two issues: > 1) Currently if a broker received StopReplicaRequest with delete=true for the > same offline replica, the first StopRelicaRequest will show > KafkaStorageException and the second StopRelicaRequest will show > ReplicaNotAvailableException. This is because the first StopRelicaRequest > will remove the mapping (tp -> ReplicaManager.OfflinePartition) from > ReplicaManager.allPartitions before returning KafkaStorageException, thus the > second StopRelicaRequest will not find this partition as offline. > This result appears to be inconsistent. And since the replica is already > offline and broker will not be able to delete file for this replica, the > StopReplicaRequest should fail without making any change and broker should > still remember that this replica is offline. > 2) Currently if broker receives StopReplicaRequest with delete=true, the > broker will attempt to remove future replica for the partition, which will > cause KafkaStorageException in the StopReplicaResponse if this replica does > not have future replica. It is problematic to always return > KafkaStorageException in the response if future replica does not exist. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7313) KAFKA-7313; StopReplicaRequest should attempt to remove future replica for the partition only if future replica exists
[ https://issues.apache.org/jira/browse/KAFKA-7313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dong Lin updated KAFKA-7313: Description: This patch fixes two issues: 1) Currently if a broker received StopReplicaRequest with delete=true for the same offline replica, the first StopRelicaRequest will show KafkaStorageException and the second StopRelicaRequest will show ReplicaNotAvailableException. This is because the first StopRelicaRequest will remove the mapping (tp -> ReplicaManager.OfflinePartition) from ReplicaManager.allPartitions before returning KafkaStorageException, thus the second StopRelicaRequest will not find this partition as offline. This result appears to be inconsistent. And since the replica is already offline and broker will not be able to delete file for this replica, the StopReplicaRequest should fail without making any change and broker should still remember that this replica is offline. 2) Currently if broker receives StopReplicaRequest with delete=true, the broker will attempt to remove future replica for the partition, which will cause KafkaStorageException in the StopReplicaResponse if this replica does not have future replica. It is problematic to always return KafkaStorageException in the response if future replica does not exist. was: Currently if a broker received StopReplicaRequest with delete=true for the same offline replica, the first StopRelicaRequest will show KafkaStorageException and the second StopRelicaRequest will show ReplicaNotAvailableException. This is because the first StopRelicaRequest will remove the mapping (tp -> ReplicaManager.OfflinePartition) from ReplicaManager.allPartitions before returning KafkaStorageException, thus the second StopRelicaRequest will not find this partition as offline. This result appears to be inconsistent. And since the replica is already offline and broker will not be able to delete file for this replica, the StopReplicaRequest should fail without making any change and broker should still remember that this replica is offline. > KAFKA-7313; StopReplicaRequest should attempt to remove future replica for > the partition only if future replica exists > -- > > Key: KAFKA-7313 > URL: https://issues.apache.org/jira/browse/KAFKA-7313 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > > This patch fixes two issues: > > 1) Currently if a broker received StopReplicaRequest with delete=true for the > same offline replica, the first StopRelicaRequest will show > KafkaStorageException and the second StopRelicaRequest will show > ReplicaNotAvailableException. This is because the first StopRelicaRequest > will remove the mapping (tp -> ReplicaManager.OfflinePartition) from > ReplicaManager.allPartitions before returning KafkaStorageException, thus the > second StopRelicaRequest will not find this partition as offline. > This result appears to be inconsistent. And since the replica is already > offline and broker will not be able to delete file for this replica, the > StopReplicaRequest should fail without making any change and broker should > still remember that this replica is offline. > > 2) Currently if broker receives StopReplicaRequest with delete=true, the > broker will attempt to remove future replica for the partition, which will > cause KafkaStorageException in the StopReplicaResponse if this replica does > not have future replica. It is problematic to always return > KafkaStorageException in the response if future replica does not exist. > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7313) KAFKA-7313; StopReplicaRequest should attempt to remove future replica for the partition only if future replica exists
[ https://issues.apache.org/jira/browse/KAFKA-7313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dong Lin updated KAFKA-7313: Summary: KAFKA-7313; StopReplicaRequest should attempt to remove future replica for the partition only if future replica exists (was: StopReplicaRequest should not remove partition from ReplicaManager.allPartitions if the replica is offline) > KAFKA-7313; StopReplicaRequest should attempt to remove future replica for > the partition only if future replica exists > -- > > Key: KAFKA-7313 > URL: https://issues.apache.org/jira/browse/KAFKA-7313 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > > Currently if a broker received StopReplicaRequest with delete=true for the > same offline replica, the first StopRelicaRequest will show > KafkaStorageException and the second StopRelicaRequest will show > ReplicaNotAvailableException. This is because the first StopRelicaRequest > will remove the mapping (tp -> ReplicaManager.OfflinePartition) from > ReplicaManager.allPartitions before returning KafkaStorageException, thus the > second StopRelicaRequest will not find this partition as offline. > This result appears to be inconsistent. And since the replica is already > offline and broker will not be able to delete file for this replica, the > StopReplicaRequest should fail without making any change and broker should > still remember that this replica is offline. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (KAFKA-6835) Enable topic unclean leader election to be enabled without controller change
[ https://issues.apache.org/jira/browse/KAFKA-6835?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jun Rao resolved KAFKA-6835. Resolution: Fixed Merged the PR to trunk. > Enable topic unclean leader election to be enabled without controller change > > > Key: KAFKA-6835 > URL: https://issues.apache.org/jira/browse/KAFKA-6835 > Project: Kafka > Issue Type: Task > Components: core >Reporter: Rajini Sivaram >Assignee: Manikumar >Priority: Major > Fix For: 2.1.0 > > > Dynamic update of broker's default unclean.leader.election.enable will be > processed without controller change (KAFKA-6526). We should probably do the > same for topic overrides as well. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-6835) Enable topic unclean leader election to be enabled without controller change
[
https://issues.apache.org/jira/browse/KAFKA-6835?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16586204#comment-16586204
]
ASF GitHub Bot commented on KAFKA-6835:
---
junrao closed pull request #4957: KAFKA-6835: Enable topic unclean leader
election to be enabled without controller change
URL: https://github.com/apache/kafka/pull/4957
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/scala/kafka/controller/ControllerState.scala
b/core/src/main/scala/kafka/controller/ControllerState.scala
index d2473058ac7..aa41c7f5458 100644
--- a/core/src/main/scala/kafka/controller/ControllerState.scala
+++ b/core/src/main/scala/kafka/controller/ControllerState.scala
@@ -94,7 +94,11 @@ object ControllerState {
def value = 13
}
+ case object TopicUncleanLeaderElectionEnable extends ControllerState {
+def value = 14
+ }
+
val values: Seq[ControllerState] = Seq(Idle, ControllerChange, BrokerChange,
TopicChange, TopicDeletion,
PartitionReassignment, AutoLeaderBalance, ManualLeaderBalance,
ControlledShutdown, IsrChange, LeaderAndIsrResponseReceived,
-LogDirChange, ControllerShutdown, UncleanLeaderElectionEnable)
+LogDirChange, ControllerShutdown, UncleanLeaderElectionEnable,
TopicUncleanLeaderElectionEnable)
}
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala
b/core/src/main/scala/kafka/controller/KafkaController.scala
index e397e80db66..f6ea43da062 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -202,6 +202,12 @@ class KafkaController(val config: KafkaConfig, zkClient:
KafkaZkClient, time: Ti
eventManager.put(UncleanLeaderElectionEnable)
}
+ private[kafka] def enableTopicUncleanLeaderElection(topic: String): Unit = {
+if (isActive) {
+ eventManager.put(TopicUncleanLeaderElectionEnable(topic))
+}
+ }
+
private def state: ControllerState = eventManager.state
/**
@@ -1025,6 +1031,16 @@ class KafkaController(val config: KafkaConfig, zkClient:
KafkaZkClient, time: Ti
}
}
+ case class TopicUncleanLeaderElectionEnable(topic: String) extends
ControllerEvent {
+
+def state = ControllerState.TopicUncleanLeaderElectionEnable
+
+override def process(): Unit = {
+ if (!isActive) return
+ partitionStateMachine.triggerOnlinePartitionStateChange(topic)
+}
+ }
+
case class ControlledShutdown(id: Int, controlledShutdownCallback:
Try[Set[TopicPartition]] => Unit) extends ControllerEvent {
def state = ControllerState.ControlledShutdown
diff --git a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
index db4c7161f35..11e38d46ffd 100755
--- a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
+++ b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
@@ -97,6 +97,14 @@ class PartitionStateMachine(config: KafkaConfig,
* state. This is called on a successful controller election and on broker
changes
*/
def triggerOnlinePartitionStateChange() {
+triggerOnlinePartitionStateChange(partitionState.toMap)
+ }
+
+ def triggerOnlinePartitionStateChange(topic: String) {
+triggerOnlinePartitionStateChange(partitionState.filterKeys(p =>
p.topic.equals(topic)).toMap)
+ }
+
+ def triggerOnlinePartitionStateChange(partitionState: Map[TopicPartition,
PartitionState]) {
// try to move all partitions in NewPartition or OfflinePartition state to
OnlinePartition state except partitions
// that belong to topics to be deleted
val partitionsToTrigger = partitionState.filter { case (partition,
partitionState) =>
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala
b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 78c3abf164a..5593225f0e6 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -21,6 +21,7 @@ import java.util.Properties
import DynamicConfig.Broker._
import kafka.api.ApiVersion
+import kafka.controller.KafkaController
import kafka.log.{LogConfig, LogManager}
import kafka.security.CredentialProvider
import kafka.server.Constants._
@@ -33,6 +34,7 @@ import org.apache.kafka.common.metrics.Quota._
import org.apache.kafka.common.utils.Sanitizer
import scala.collection.JavaConverters._
+import scala.util.Try
/**
* The ConfigHandler is used to process config change notifications received
by the DynamicConfigManager
@@ -45,7 +47,7 @@ trait ConfigHandler {
* The TopicConfigHandler will process topic config changes in ZK.
* The callback provides
[jira] [Resolved] (KAFKA-7210) Add system test for log compaction
[ https://issues.apache.org/jira/browse/KAFKA-7210?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma resolved KAFKA-7210. Resolution: Fixed > Add system test for log compaction > -- > > Key: KAFKA-7210 > URL: https://issues.apache.org/jira/browse/KAFKA-7210 > Project: Kafka > Issue Type: Improvement >Reporter: Manikumar >Assignee: Manikumar >Priority: Major > Fix For: 2.1.0 > > > Currently we have TestLogCleaning tool for stress test log compaction. This > JIRA is to integrate the tool to system test. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7210) Add system test for log compaction
[
https://issues.apache.org/jira/browse/KAFKA-7210?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16585776#comment-16585776
]
ASF GitHub Bot commented on KAFKA-7210:
---
ijuma closed pull request #5226: KAFKA-7210: Add a system test to verify the
log compaction
URL: https://github.com/apache/kafka/pull/5226
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/build.gradle b/build.gradle
index 0892ed19402..c1387d4ef60 100644
--- a/build.gradle
+++ b/build.gradle
@@ -775,6 +775,9 @@ project(':core') {
include('*.jar')
}
into "$buildDir/dependant-testlibs"
+//By default gradle does not handle test dependencies between the
sub-projects
+//This line is to include clients project test jar to dependant-testlibs
+from (project(':clients').testJar ) { "$buildDir/dependant-testlibs" }
duplicatesStrategy 'exclude'
}
diff --git a/core/src/test/scala/kafka/tools/LogCompactionTester.scala
b/core/src/test/scala/kafka/tools/LogCompactionTester.scala
new file mode 100755
index 000..9f53f664b54
--- /dev/null
+++ b/core/src/test/scala/kafka/tools/LogCompactionTester.scala
@@ -0,0 +1,348 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.tools
+
+import java.io._
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.{Files, Path}
+import java.time.Duration
+import java.util.{Properties, Random}
+
+import joptsimple.OptionParser
+import kafka.utils._
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.{CommonClientConfigs, admin}
+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig,
ProducerRecord}
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.serialization.{ByteArraySerializer,
StringDeserializer}
+import org.apache.kafka.common.utils.Utils
+
+import scala.collection.JavaConverters._
+
+/**
+ * This is a torture test that runs against an existing broker
+ *
+ * Here is how it works:
+ *
+ * It produces a series of specially formatted messages to one or more
partitions. Each message it produces
+ * it logs out to a text file. The messages have a limited set of keys, so
there is duplication in the key space.
+ *
+ * The broker will clean its log as the test runs.
+ *
+ * When the specified number of messages have been produced we create a
consumer and consume all the messages in the topic
+ * and write that out to another text file.
+ *
+ * Using a stable unix sort we sort both the producer log of what was sent and
the consumer log of what was retrieved by the message key.
+ * Then we compare the final message in both logs for each key. If this final
message is not the same for all keys we
+ * print an error and exit with exit code 1, otherwise we print the size
reduction and exit with exit code 0.
+ */
+object LogCompactionTester {
+
+ //maximum line size while reading produced/consumed record text file
+ private val ReadAheadLimit = 4906
+
+ def main(args: Array[String]) {
+val parser = new OptionParser(false)
+val numMessagesOpt = parser.accepts("messages", "The number of messages to
send or consume.")
+ .withRequiredArg
+ .describedAs("count")
+ .ofType(classOf[java.lang.Long])
+ .defaultsTo(Long.MaxValue)
+val messageCompressionOpt = parser.accepts("compression-type", "message
compression type")
+ .withOptionalArg
+ .describedAs("compressionType")
+ .ofType(classOf[java.lang.String])
+ .defaultsTo("none")
+val numDupsOpt = parser.accepts("duplicates", "The number of duplicates
for each key.")
+ .withRequiredArg
+ .describedAs("count")
+ .ofType(classOf[java.lang.Integer])
+ .defaultsTo(5)
+val brokerOpt = parser.accepts("bootstrap-server", "The server(s) to
connect to.")
+ .withRequiredArg
+
[jira] [Commented] (KAFKA-4740) Using new consumer API with a Deserializer that throws SerializationException can lead to infinite loop
[
https://issues.apache.org/jira/browse/KAFKA-4740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16585748#comment-16585748
]
Julien Debon commented on KAFKA-4740:
-
[~rocketraman] From the look of it, it is not possible today to seek the next
offset, as you don't know which partition caused the issue.
As [~hachikuji] mentioned in the second comment, this piece of information is
currently not available. An option would be to make it available in the
exception.
> Using new consumer API with a Deserializer that throws SerializationException
> can lead to infinite loop
> ---
>
> Key: KAFKA-4740
> URL: https://issues.apache.org/jira/browse/KAFKA-4740
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Affects Versions: 0.9.0.0, 0.9.0.1, 0.10.0.0, 0.10.0.1, 0.10.1.0, 0.10.1.1
> Environment: Kafka broker 0.10.1.1 (but this bug is not dependent on
> the broker version)
> Kafka clients 0.9.0.0, 0.9.0.1, 0.10.0.0, 0.10.0.1, 0.10.1.0, 0.10.1.1
>Reporter: Sébastien Launay
>Assignee: Sébastien Launay
>Priority: Critical
>
> The old consumer supports deserializing records into typed objects and throws
> a {{SerializationException}} through {{MessageAndMetadata#key()}} and
> {{MessageAndMetadata#message()}} that can be catched by the client \[1\].
> When using the new consumer API with kafka-clients version < 0.10.0.1, such
> the exception is swallowed by the {{NetworkClient}} class and result in an
> infinite loop which the client has no control over like:
> {noformat}
> DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - Resetting offset
> for partition test2-0 to earliest offset.
> DEBUG org.apache.kafka.clients.consumer.internals.Fetcher - Fetched offset 0
> for partition test2-0
> ERROR org.apache.kafka.clients.NetworkClient - Uncaught error in request
> completion:
> org.apache.kafka.common.errors.SerializationException: Size of data received
> by IntegerDeserializer is not 4
> ERROR org.apache.kafka.clients.NetworkClient - Uncaught error in request
> completion:
> org.apache.kafka.common.errors.SerializationException: Size of data received
> by IntegerDeserializer is not 4
> ...
> {noformat}
> Thanks to KAFKA-3977, this has been partially fixed in 0.10.1.0 but another
> issue still remains.
> Indeed, the client can now catch the {{SerializationException}} but the next
> call to {{Consumer#poll(long)}} will throw the same exception indefinitely.
> The following snippet (full example available on Github \[2\] for most
> released kafka-clients versions):
> {code:java}
> try (KafkaConsumer kafkaConsumer = new
> KafkaConsumer<>(consumerConfig, new StringDeserializer(), new
> IntegerDeserializer())) {
> kafkaConsumer.subscribe(Arrays.asList("topic"));
> // Will run till the shutdown hook is called
> while (!doStop) {
> try {
> ConsumerRecords records =
> kafkaConsumer.poll(1000);
> if (!records.isEmpty()) {
> logger.info("Got {} messages", records.count());
> for (ConsumerRecord record : records) {
> logger.info("Message with partition: {}, offset: {}, key:
> {}, value: {}",
> record.partition(), record.offset(), record.key(),
> record.value());
> }
> } else {
> logger.info("No messages to consume");
> }
> } catch (SerializationException e) {
> logger.warn("Failed polling some records", e);
> }
> }
> }
> {code}
> when run with the following records (third record has an invalid Integer
> value):
> {noformat}
> printf "\x00\x00\x00\x00\n" | bin/kafka-console-producer.sh --broker-list
> localhost:9092 --topic topic
> printf "\x00\x00\x00\x01\n" | bin/kafka-console-producer.sh --broker-list
> localhost:9092 --topic topic
> printf "\x00\x00\x00\n" | bin/kafka-console-producer.sh --broker-list
> localhost:9092 --topic topic
> printf "\x00\x00\x00\x02\n" | bin/kafka-console-producer.sh --broker-list
> localhost:9092 --topic topic
> {noformat}
> will produce the following logs:
> {noformat}
> INFO consumer.Consumer - Got 2 messages
> INFO consumer.Consumer - Message with partition: 0, offset: 0, key: null,
> value: 0
> INFO consumer.Consumer - Message with partition: 0, offset: 1, key: null,
> value: 1
> WARN consumer.Consumer - Failed polling some records
> org.apache.kafka.common.errors.SerializationException: Error deserializing
> key/value for partition topic-0 at offset 2
> Caused by: org.apache.kafka.common.errors.SerializationException: Size of
> data received by IntegerDeserializer is not 4
> WARN consumer.Consumer - Failed polling some records
>
[jira] [Commented] (KAFKA-960) Upgrade Metrics to 3.x
[ https://issues.apache.org/jira/browse/KAFKA-960?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16585713#comment-16585713 ] Erik van Oosten commented on KAFKA-960: --- Metrics 4.x was released not so long ago. The core is binary compatible with metrics 3.x. However, many modules were split from the core and these got a different package name (and are therefore not compatible). For just collecting, you're probably fine. Please also know that Metrics 5.x is on standby for more then half a year. Metrics 5 will support tags. Metrics 5 is not binary compatible. I recommend upgrading to Metrics 4. > Upgrade Metrics to 3.x > -- > > Key: KAFKA-960 > URL: https://issues.apache.org/jira/browse/KAFKA-960 > Project: Kafka > Issue Type: Improvement > Components: metrics >Affects Versions: 0.8.1 >Reporter: Cosmin Lehene >Priority: Major > > Now that metrics 3.0 has been released > (http://metrics.codahale.com/about/release-notes/) we can upgrade back -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-7313) StopReplicaRequest should not remove partition from ReplicaManager.allPartitions if the replica is offline
[ https://issues.apache.org/jira/browse/KAFKA-7313?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16585513#comment-16585513 ] ASF GitHub Bot commented on KAFKA-7313: --- lindong28 opened a new pull request #5533: KAFKA-7313; StopReplicaRequest should not remove partition from ReplicaManager.allPartitions if the replica is offline URL: https://github.com/apache/kafka/pull/5533 ### 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 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 > StopReplicaRequest should not remove partition from > ReplicaManager.allPartitions if the replica is offline > -- > > Key: KAFKA-7313 > URL: https://issues.apache.org/jira/browse/KAFKA-7313 > Project: Kafka > Issue Type: Improvement >Reporter: Dong Lin >Assignee: Dong Lin >Priority: Major > > Currently if a broker received StopReplicaRequest with delete=true for the > same offline replica, the first StopRelicaRequest will show > KafkaStorageException and the second StopRelicaRequest will show > ReplicaNotAvailableException. This is because the first StopRelicaRequest > will remove the mapping (tp -> ReplicaManager.OfflinePartition) from > ReplicaManager.allPartitions before returning KafkaStorageException, thus the > second StopRelicaRequest will not find this partition as offline. > This result appears to be inconsistent. And since the replica is already > offline and broker will not be able to delete file for this replica, the > StopReplicaRequest should fail without making any change and broker should > still remember that this replica is offline. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
