Re: Issue in Kafka 2.0.0 ?

[Ted Yu]

Sent out a PR #5543 which fixes the reported bug,
with 
StreamToTableJoinScalaIntegrationTestImplicitSerdes.testShouldCountClicksPerRegion
modified adding the filter methods.

FYI

On Mon, Aug 20, 2018 at 5:26 PM Ted Yu  wrote:

> Thanks for pointing me to that PR.
>
> I applied the PR locally but still got:
>
> org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
> > testShouldCountClicksPerRegion FAILED
> java.lang.StackOverflowError
>
> I can go over that PR to see what can be referenced for solving this bug.
>
> FYI
>
> On Mon, Aug 20, 2018 at 5:21 PM Guozhang Wang  wrote:
>
>> Is this related to the fix https://github.com/apache/kafka/pull/5502 that
>> is currently being worked on?
>>
>>
>> Guozhang
>>
>> On Mon, Aug 20, 2018 at 5:19 PM, Matthias J. Sax 
>> wrote:
>>
>> > Thanks for reporting and for creating the ticket!
>> >
>> > -Matthias
>> >
>> > On 8/20/18 5:17 PM, Ted Yu wrote:
>> > > I was able to reproduce what you saw with modification
>> > > to StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
>> > > I have logged KAFKA-7316 and am looking for a fix.
>> > >
>> > > FYI
>> > >
>> > > On Mon, Aug 20, 2018 at 1:39 PM Druhin Sagar Goel 
>> > wrote:
>> > >
>> > >> Isn’t that a bug then? Or can I fix my code somehow?
>> > >>
>> > >>
>> > >>
>> > >> On August 20, 2018 at 1:30:42 PM, Ted Yu (yuzhih...@gmail.com
>> > > >> yuzhih...@gmail.com>) wrote:
>> > >>
>> > >> I think what happened in your use case was that the following
>> implicit
>> > >> from ImplicitConversions.scala kept wrapping the resultant KTable
>> from
>> > >> filter():
>> > >>
>> > >> implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =
>> > >>
>> > >> leading to stack overflow.
>> > >>
>> > >> Cheers
>> > >>
>> > >> On Mon, Aug 20, 2018 at 12:50 PM Druhin Sagar Goel <
>> dru...@arrcus.com>
>> > >> wrote:
>> > >>
>> > >>> Hi,
>> > >>>
>> > >>> I’m using the org.kafka.streams.scala that was released with version
>> > >>> 2.0.0. I’m getting a StackOverflowError as follows:
>> > >>>
>> > >>> java.lang.StackOverflowError
>> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
>> > KTable.scala:49)
>> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
>> > KTable.scala:49)
>> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
>> > KTable.scala:49)
>> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
>> > KTable.scala:49)
>> > >>> .
>> > >>> .
>> > >>> .
>> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
>> > KTable.scala:49)
>> > >>>
>> > >>> The Scala version I’m using is 2.11.11 and the code leading to the
>> > error
>> > >>> is as follows (particularly the .filter).
>> > >>>
>> > >>> val builder = new StreamsBuilder
>> > >>>
>> > >>> val stream = builder.stream[Array[Byte], CaseClassA](args.topic)
>> > >>>
>> > >>> val customers = args.config.keys
>> > >>>
>> > >>> val predicates = customers.map { customerId =>
>> > >>> (_: Array[Byte], message: CaseClassA) => message.customerId ==
>> > customerId
>> > >>> }.toSeq
>> > >>>
>> > >>> val customerIdToStream = customers.zip(stream(predicates: _*)).toMap
>> > >>>
>> > >>> val y = Printed.toSysOut[Windowed[Key], Long]
>> > >>>
>> > >>> customerIdToStream.foreach { case (customerId, customerStream) =>
>> > >>> val customerConfig = args.config(customerId)
>> > >>> customerStream
>> > >>> .flatMap { case (_, message) =>
>> > >>> message.objects.map {
>> > >>> case CaseClassB(c, _) => Key(message.id, c.prefix) -> 1
>> > >>> }
>> > >>> }
>> > >>> .groupByKey
>> > >>>
>> > >>>
>> > >> .windowedBy(TimeWindows.of(customerConfig.windowSize).
>> > advanceBy(customerConfig.sliderSize))
>> > >>> .count()
>> > >>> .filter { case (_, count) => count >=
>> > >>> customerConfig.frequencyThreshold }
>> > >>> .toStream
>> > >>> .print(y)
>> > >>> }
>> > >>>
>> > >>> Is this a bug with the new scala module related to:
>> > >>> https://github.com/lightbend/kafka-streams-scala/issues/63 ?
>> > >>> Or am I doing something wrong?
>> > >>>
>> > >>> Thanks,
>> > >>> Druhin
>> > >>>
>> > >>
>> > >
>> >
>> >
>>
>>
>> --
>> -- Guozhang
>>
>

Re: Issue in Kafka 2.0.0 ?

[Ted Yu]

Hi,
I am aware that more than one method from KTable.scala have this issue.

Once I find a solution, I will apply the fix to the methods you listed.

Cheers

On Mon, Aug 20, 2018 at 5:23 PM Druhin Sagar Goel  wrote:

> Thanks a lot Ted!
>
> FYI - The issue is not limited to the
> org.apache.kafka.streams.scala.KTable.filter. It also happens with
> org.apache.kafka.streams.scala.KTable.filterNot,
> org.apache.kafka.streams.scala.KStream.foreach and
> org.apache.kafka.streams.scala.KStream.peek.
>
> - Druhin
>
>
> On August 20, 2018 at 5:19:36 PM, Matthias J. Sax (matth...@confluent.io
> <mailto:matth...@confluent.io>) wrote:
>
> Thanks for reporting and for creating the ticket!
>
> -Matthias
>
> On 8/20/18 5:17 PM, Ted Yu wrote:
> > I was able to reproduce what you saw with modification
> > to StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> > I have logged KAFKA-7316 and am looking for a fix.
> >
> > FYI
> >
> > On Mon, Aug 20, 2018 at 1:39 PM Druhin Sagar Goel 
> wrote:
> >
> >> Isn’t that a bug then? Or can I fix my code somehow?
> >>
> >>
> >>
> >> On August 20, 2018 at 1:30:42 PM, Ted Yu (yuzhih...@gmail.com >> yuzhih...@gmail.com>) wrote:
> >>
> >> I think what happened in your use case was that the following implicit
> >> from ImplicitConversions.scala kept wrapping the resultant KTable from
> >> filter():
> >>
> >> implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =
> >>
> >> leading to stack overflow.
> >>
> >> Cheers
> >>
> >> On Mon, Aug 20, 2018 at 12:50 PM Druhin Sagar Goel 
> >> wrote:
> >>
> >>> Hi,
> >>>
> >>> I’m using the org.kafka.streams.scala that was released with version
> >>> 2.0.0. I’m getting a StackOverflowError as follows:
> >>>
> >>> java.lang.StackOverflowError
> >>> at
> org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >>> at
> org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >>> at
> org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >>> at
> org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >>> .
> >>> .
> >>> .
> >>> at
> org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >>>
> >>> The Scala version I’m using is 2.11.11 and the code leading to the
> error
> >>> is as follows (particularly the .filter).
> >>>
> >>> val builder = new StreamsBuilder
> >>>
> >>> val stream = builder.stream[Array[Byte], CaseClassA](args.topic)
> >>>
> >>> val customers = args.config.keys
> >>>
> >>> val predicates = customers.map { customerId =>
> >>> (_: Array[Byte], message: CaseClassA) => message.customerId ==
> customerId
> >>> }.toSeq
> >>>
> >>> val customerIdToStream = customers.zip(stream(predicates: _*)).toMap
> >>>
> >>> val y = Printed.toSysOut[Windowed[Key], Long]
> >>>
> >>> customerIdToStream.foreach { case (customerId, customerStream) =>
> >>> val customerConfig = args.config(customerId)
> >>> customerStream
> >>> .flatMap { case (_, message) =>
> >>> message.objects.map {
> >>> case CaseClassB(c, _) => Key(message.id, c.prefix) -> 1
> >>> }
> >>> }
> >>> .groupByKey
> >>>
> >>>
> >>
> .windowedBy(TimeWindows.of(customerConfig.windowSize).advanceBy(customerConfig.sliderSize))
> >>> .count()
> >>> .filter { case (_, count) => count >=
> >>> customerConfig.frequencyThreshold }
> >>> .toStream
> >>> .print(y)
> >>> }
> >>>
> >>> Is this a bug with the new scala module related to:
> >>> https://github.com/lightbend/kafka-streams-scala/issues/63 ?
> >>> Or am I doing something wrong?
> >>>
> >>> Thanks,
> >>> Druhin
> >>>
> >>
> >
>
>

Re: Issue in Kafka 2.0.0 ?

[Ted Yu]

Thanks for pointing me to that PR.

I applied the PR locally but still got:

org.apache.kafka.streams.scala.StreamToTableJoinScalaIntegrationTestImplicitSerdes
> testShouldCountClicksPerRegion FAILED
java.lang.StackOverflowError

I can go over that PR to see what can be referenced for solving this bug.

FYI

On Mon, Aug 20, 2018 at 5:21 PM Guozhang Wang  wrote:

> Is this related to the fix https://github.com/apache/kafka/pull/5502 that
> is currently being worked on?
>
>
> Guozhang
>
> On Mon, Aug 20, 2018 at 5:19 PM, Matthias J. Sax 
> wrote:
>
> > Thanks for reporting and for creating the ticket!
> >
> > -Matthias
> >
> > On 8/20/18 5:17 PM, Ted Yu wrote:
> > > I was able to reproduce what you saw with modification
> > > to StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
> > > I have logged KAFKA-7316 and am looking for a fix.
> > >
> > > FYI
> > >
> > > On Mon, Aug 20, 2018 at 1:39 PM Druhin Sagar Goel 
> > wrote:
> > >
> > >> Isn’t that a bug then? Or can I fix my code somehow?
> > >>
> > >>
> > >>
> > >> On August 20, 2018 at 1:30:42 PM, Ted Yu (yuzhih...@gmail.com > >> yuzhih...@gmail.com>) wrote:
> > >>
> > >> I think what happened in your use case was that the following implicit
> > >> from ImplicitConversions.scala kept wrapping the resultant KTable from
> > >> filter():
> > >>
> > >> implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =
> > >>
> > >> leading to stack overflow.
> > >>
> > >> Cheers
> > >>
> > >> On Mon, Aug 20, 2018 at 12:50 PM Druhin Sagar Goel  >
> > >> wrote:
> > >>
> > >>> Hi,
> > >>>
> > >>> I’m using the org.kafka.streams.scala that was released with version
> > >>> 2.0.0. I’m getting a StackOverflowError as follows:
> > >>>
> > >>> java.lang.StackOverflowError
> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
> > KTable.scala:49)
> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
> > KTable.scala:49)
> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
> > KTable.scala:49)
> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
> > KTable.scala:49)
> > >>> .
> > >>> .
> > >>> .
> > >>> at org.apache.kafka.streams.scala.kstream.KTable.filter(
> > KTable.scala:49)
> > >>>
> > >>> The Scala version I’m using is 2.11.11 and the code leading to the
> > error
> > >>> is as follows (particularly the .filter).
> > >>>
> > >>> val builder = new StreamsBuilder
> > >>>
> > >>> val stream = builder.stream[Array[Byte], CaseClassA](args.topic)
> > >>>
> > >>> val customers = args.config.keys
> > >>>
> > >>> val predicates = customers.map { customerId =>
> > >>> (_: Array[Byte], message: CaseClassA) => message.customerId ==
> > customerId
> > >>> }.toSeq
> > >>>
> > >>> val customerIdToStream = customers.zip(stream(predicates: _*)).toMap
> > >>>
> > >>> val y = Printed.toSysOut[Windowed[Key], Long]
> > >>>
> > >>> customerIdToStream.foreach { case (customerId, customerStream) =>
> > >>> val customerConfig = args.config(customerId)
> > >>> customerStream
> > >>> .flatMap { case (_, message) =>
> > >>> message.objects.map {
> > >>> case CaseClassB(c, _) => Key(message.id, c.prefix) -> 1
> > >>> }
> > >>> }
> > >>> .groupByKey
> > >>>
> > >>>
> > >> .windowedBy(TimeWindows.of(customerConfig.windowSize).
> > advanceBy(customerConfig.sliderSize))
> > >>> .count()
> > >>> .filter { case (_, count) => count >=
> > >>> customerConfig.frequencyThreshold }
> > >>> .toStream
> > >>> .print(y)
> > >>> }
> > >>>
> > >>> Is this a bug with the new scala module related to:
> > >>> https://github.com/lightbend/kafka-streams-scala/issues/63 ?
> > >>> Or am I doing something wrong?
> > >>>
> > >>> Thanks,
> > >>> Druhin
> > >>>
> > >>
> > >
> >
> >
>
>
> --
> -- Guozhang
>

Re: Issue in Kafka 2.0.0 ?

[Ted Yu]

I was able to reproduce what you saw with modification
to StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
I have logged KAFKA-7316 and am looking for a fix.

FYI

On Mon, Aug 20, 2018 at 1:39 PM Druhin Sagar Goel  wrote:

> Isn’t that a bug then? Or can I fix my code somehow?
>
>
>
> On August 20, 2018 at 1:30:42 PM, Ted Yu (yuzhih...@gmail.com yuzhih...@gmail.com>) wrote:
>
> I think what happened in your use case was that the following implicit
> from ImplicitConversions.scala kept wrapping the resultant KTable from
> filter():
>
> implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =
>
> leading to stack overflow.
>
> Cheers
>
> On Mon, Aug 20, 2018 at 12:50 PM Druhin Sagar Goel 
> wrote:
>
> > Hi,
> >
> > I’m using the org.kafka.streams.scala that was released with version
> > 2.0.0. I’m getting a StackOverflowError as follows:
> >
> > java.lang.StackOverflowError
> > at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> > at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> > at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> > at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> > .
> > .
> > .
> > at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> >
> > The Scala version I’m using is 2.11.11 and the code leading to the error
> > is as follows (particularly the .filter).
> >
> > val builder = new StreamsBuilder
> >
> > val stream = builder.stream[Array[Byte], CaseClassA](args.topic)
> >
> > val customers = args.config.keys
> >
> > val predicates = customers.map { customerId =>
> > (_: Array[Byte], message: CaseClassA) => message.customerId == customerId
> > }.toSeq
> >
> > val customerIdToStream = customers.zip(stream(predicates: _*)).toMap
> >
> > val y = Printed.toSysOut[Windowed[Key], Long]
> >
> > customerIdToStream.foreach { case (customerId, customerStream) =>
> > val customerConfig = args.config(customerId)
> > customerStream
> > .flatMap { case (_, message) =>
> > message.objects.map {
> > case CaseClassB(c, _) => Key(message.id, c.prefix) -> 1
> > }
> > }
> > .groupByKey
> >
> >
> .windowedBy(TimeWindows.of(customerConfig.windowSize).advanceBy(customerConfig.sliderSize))
> > .count()
> > .filter { case (_, count) => count >=
> > customerConfig.frequencyThreshold }
> > .toStream
> > .print(y)
> > }
> >
> > Is this a bug with the new scala module related to:
> > https://github.com/lightbend/kafka-streams-scala/issues/63 ?
> > Or am I doing something wrong?
> >
> > Thanks,
> > Druhin
> >
>

Re: Issue in Kafka 2.0.0 ?

[Ted Yu]

I think what happened in your use case was that the following implicit
from ImplicitConversions.scala kept wrapping the resultant KTable from
filter():

  implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =

leading to stack overflow.

Cheers

On Mon, Aug 20, 2018 at 12:50 PM Druhin Sagar Goel 
wrote:

> Hi,
>
> I’m using the org.kafka.streams.scala that was released with version
> 2.0.0. I’m getting a StackOverflowError as follows:
>
> java.lang.StackOverflowError
> at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
> at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
>.
>.
>.
> at org.apache.kafka.streams.scala.kstream.KTable.filter(KTable.scala:49)
>
> The Scala version I’m using is 2.11.11 and the code leading to the error
> is as follows (particularly the .filter).
>
> val builder = new StreamsBuilder
>
> val stream = builder.stream[Array[Byte], CaseClassA](args.topic)
>
> val customers = args.config.keys
>
> val predicates = customers.map { customerId =>
>   (_: Array[Byte], message: CaseClassA) => message.customerId == customerId
> }.toSeq
>
> val customerIdToStream = customers.zip(stream(predicates: _*)).toMap
>
> val y = Printed.toSysOut[Windowed[Key], Long]
>
> customerIdToStream.foreach { case (customerId, customerStream) =>
>   val customerConfig = args.config(customerId)
>   customerStream
> .flatMap { case (_, message) =>
>   message.objects.map {
> case CaseClassB(c, _) => Key(message.id, c.prefix) -> 1
>   }
> }
> .groupByKey
>
> .windowedBy(TimeWindows.of(customerConfig.windowSize).advanceBy(customerConfig.sliderSize))
> .count()
> .filter { case (_, count) => count >=
> customerConfig.frequencyThreshold }
> .toStream
> .print(y)
> }
>
> Is this a bug with the new scala module related to:
> https://github.com/lightbend/kafka-streams-scala/issues/63 ?
> Or am I doing something wrong?
>
> Thanks,
> Druhin
>

Re: Java API to read metrics via JMX

[Ted Yu]

Boris:
BrokerWithJMX is referenced but I didn't find the class source after a
brief search.

FYI

On Wed, Aug 8, 2018 at 7:10 PM Boris Lublinsky <
boris.lublin...@lightbend.com> wrote:

> Its actually quite simple, unfortunately you have to read, and then write
> to TSDB.
> Enclosed is an example doing this and dumping to InfluxDB
>
>
> Boris Lublinsky
> FDP Architect
> boris.lublin...@lightbend.com
> https://www.lightbend.com/
>
> On Aug 8, 2018, at 8:46 PM, Raghav  wrote:
>
> Hi
>
> Is there any Java API available so that I can enable our Kafka cluster's
> JMX port, and consume metrics via JMX api, and dump to a time series
> database.
>
> I checked out jmxtrans, but currently it does not dump to TSDB (time series
> database).
>
> Thanks.
>
> R
>
>
>

Re: Error reading field 'version' when execute kafka-consumer-groups.sh

[Ted Yu]

For the last error, please see KAFKA-6868

On Wed, Aug 8, 2018 at 8:18 AM 刘鹏  wrote:

> hi,
> I get an error when I run kafka-consumer-groups.sh.
>
>
> firstly,I run:
> ./kafka-consumer-groups.sh --command-config consumer.properties
> --bootstrap-server ip:port --list
>
> the result is:
> Note: This will not show information about old Zookeeper-based
> consumers.
> mygroup
>
> then,I run:
>./kafka-consumer-groups.sh --command-config consumer.properties
> --bootstrap-server ip:port --describe --group mygroup
>
>
>the output I get is:
>Note: This will not show information about old Zookeeper-based
> consumers.
>Error: Executing consumer group command failed due to Error reading
> field 'version': java.nio.BufferUnderflowException
>
>
>my kafka version is:2.11-1.1.0,zookeeper version is:3.4.6,I have a 5
> nodes zk cluster and a 5 node kafka brokers cluster,they are in the same 5
> machine.I am using sarama-cluster as kafka client in golang.
>
>
>How can I find more detail of the error? Is anyone have some idea when
> this error occur?
>
>
>thanks a lot.
>
>
>
>
> thank you
> liupeng
>
>

Re: Kafka flow

[Ted Yu]

Have you looked at http://kafka.apache.org/documentation/ ?

Chapters 4 and 5.

FYI

On Wed, Aug 8, 2018 at 1:46 AM darekAsz  wrote:

> Hi,
> I need informations about workflow in kafka, I want to know what is going
> on when producer send message, as much details as possible about
> requests-responses. Can anyone tell me when I can find this?
> Best regards
>

Re: Apache Kafka Process showing high CPU (100 to 200+) usage in Linux when Idle

[Ted Yu]

bq. only one specific node is showing this issue

Is controller running on this node ? updating the metrics is expensive.

Cheers

On Sat, Aug 4, 2018 at 3:00 PM Abhijith Sreenivasan <
abhijithonl...@gmail.com> wrote:

> Hello
>
> We are seeing high CPU usage for the Kafka process. I am using 0.11
> version. Has 5 topics out of which 1 was created newly. We attempted to
> publish message this new topic which did not show up in the consumer, but
> no errors in the publisher end. Not sure why the message did not show up in
> consumer.
>
> This ran for a couple of days (30K messages) when we noticed 100%+ CPU
> usage. Tried deleting the topic (config is enabled), it was marked for
> deletion but after which usage rose to below levels 240%+. We restarted the
> process many times and disabled the publisher/producer but no difference.
> After some time (1 or 2 hours) we are getting a "Too many open files" error
> and process is shutting down.
>
> We have 3 nodes with Kafka and 3 other nodes running with ZK, but only one
> specific node is showing this issue. (where new topic partition is
> present).
>
> Still debugging and this is a prod environment.. please help!
>
> Thanks,
> Abhi
>
> top - 17:47:43 up 289 days, 18:54,  2 users,  load average: 2.65, 2.72,
> 2.52
> Tasks: 144 total,   1 running, 143 sleeping,   0 stopped,   0 zombie
> %Cpu(s): 37.5 us, 19.4 sy,  0.0 ni, 37.0 id,  0.0 wa,  0.0 hi,  5.4 si,
> 0.7 st
> KiB Mem : 16266464 total,  1431916 free,  5769976 used,  9064572 buff/cache
> KiB Swap:0 total,0 free,0 used.  9230548 avail Mem
>
>   PID USER  PR  NIVIRTRESSHR S  %CPU %MEM TIME+ COMMAND
> 32058 root  20   0 5898348 1.078g  15548 S 253.0  6.9  99:03.68 java
>10 root  20   0   0  0  0 S   0.3  0.0 921:42.77
> rcu_sche
>

Re: I can't use the command to close my kafka

[Ted Yu]

Can you provide a bit more information ?

Release of Kafka

Have you checked broker logs after running kafka-server-stop.sh ?

Cheers

On Fri, Jul 27, 2018 at 3:53 PM ? ??  wrote:

> Dear apache staff,
> I'm sorry to bother you during your busy schedule, but I had a problem
> with kafka! When I started kafka on centOS-6.5, I couldn't close it with
> the kafka-server-stop.sh command. I hope my poor English will not make you
> feel impatient or unable to describe my problem clearly!
> Sincerely yours,
> LeLe
>

Re: Next Release

[Ted Yu]

Have you seen this thread ?

http://search-hadoop.com/m/Kafka/uyzND1050Xu1lWkZV?subj=+VOTE+2+0+0+RC3

On Wed, Jul 25, 2018 at 7:04 PM xiejing 
wrote:

> Hi,
>
>
>
> Do you have a schedule for the next release, say 1.2. or 2.0?
>
>
> --
>
> 谢静.Kane
>
> 大数据平台部
> [image: cid:_Foxmail.1@abc581d8-f944-5eb8-e3d2-9b6d60ae101c]
> 上海浦东新区海趣路58号1号楼
>
> 电话：021-50504740-893325
>
> 邮件：xiejing.k...@shandagames.com
>
>
> 本邮件及其附件含有盛大游戏的保密信息，仅限于发送给上面地址中列出的个人或群组。禁止任何其他人以任何形式使用（包括但不限于全部或部分地泄露、复制、或散发）本邮件中的信息。如果您错收了本邮件，请您立即电话或邮件通知发件人并删除本邮件！
>
>
>

Re: [VOTE] 2.0.0 RC3

[Ted Yu]

+1

Checked signatures

Ran test suite which passed.

On Tue, Jul 24, 2018 at 8:28 AM Rajini Sivaram 
wrote:

> Hello Kafka users, developers and client-developers,
>
>
> This is the fourth candidate for release of Apache Kafka 2.0.0.
>
>
> This is a major version release of Apache Kafka. It includes 40 new  KIPs
> and
>
> several critical bug fixes. Please see the 2.0.0 release plan for more
> details:
>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=80448820
>
>
> A few notable highlights:
>
>- Prefixed wildcard ACLs (KIP-290), Fine grained ACLs for CreateTopics
>(KIP-277)
>- SASL/OAUTHBEARER implementation (KIP-255)
>- Improved quota communication and customization of quotas (KIP-219,
>KIP-257)
>- Efficient memory usage for down conversion (KIP-283)
>- Fix log divergence between leader and follower during fast leader
>failover (KIP-279)
>- Drop support for Java 7 and remove deprecated code including old scala
>clients
>- Connect REST extension plugin, support for externalizing secrets and
>improved error handling (KIP-285, KIP-297, KIP-298 etc.)
>- Scala API for Kafka Streams and other Streams API improvements
>(KIP-270, KIP-150, KIP-245, KIP-251 etc.)
>
>
> Release notes for the 2.0.0 release:
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/RELEASE_NOTES.html
>
>
> *** Please download, test and vote by Friday July 27, 4pm PT.
>
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
>
> http://kafka.apache.org/KEYS
>
>
> * Release artifacts to be voted upon (source and binary):
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/
>
>
> * Maven artifacts to be voted upon:
>
> https://repository.apache.org/content/groups/staging/
>
>
> * Javadoc:
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/javadoc/
>
>
> * Tag to be voted upon (off 2.0 branch) is the 2.0.0 tag:
>
> https://github.com/apache/kafka/releases/tag/2.0.0-rc3
>
> * Documentation:
>
> http://kafka.apache.org/20/documentation.html
>
>
> * Protocol:
>
> http://kafka.apache.org/20/protocol.html
>
>
> * Successful Jenkins builds for the 2.0 branch:
>
> Unit/integration tests: https://builds.apache.org/job/kafka-2.0-jdk8/90/
>
> System tests:
> https://jenkins.confluent.io/job/system-test-kafka/job/2.0/41/
>
>
> /**
>
>
> Thanks,
>
>
>
> Rajini
>

Re: [VOTE] 1.1.1 RC3

[Ted Yu]

+1

Ran test suite.

Checked signatures.

On Sun, Jul 8, 2018 at 3:36 PM Dong Lin  wrote:

> Hello Kafka users, developers and client-developers,
>
> This is the fourth candidate for release of Apache Kafka 1.1.1.
>
> Apache Kafka 1.1.1 is a bug-fix release for the 1.1 branch that was first
> released with 1.1.0 about 3 months ago. We have fixed about 25 issues since
> that release. A few of the more significant fixes include:
>
> KAFKA-6925  - Fix memory
> leak in StreamsMetricsThreadImpl
> KAFKA-6937  - In-sync
> replica delayed during fetch if replica throttle is exceeded
> KAFKA-6917  - Process
> txn
> completion asynchronously to avoid deadlock
> KAFKA-6893  - Create
> processors before starting acceptor to avoid ArithmeticException
> KAFKA-6870  -
> Fix ConcurrentModificationException in SampledStat
> KAFKA-6878  - Fix
> NullPointerException when querying global state store
> KAFKA-6879  - Invoke
> session init callbacks outside lock to avoid Controller deadlock
> KAFKA-6857  - Prevent
> follower from truncating to the wrong offset if undefined leader epoch is
> requested
> KAFKA-6854  - Log
> cleaner
> fails with transaction markers that are deleted during clean
> KAFKA-6747  - Check
> whether there is in-flight transaction before aborting transaction
> KAFKA-6748  - Double
> check before scheduling a new task after the punctuate call
> KAFKA-6739  -
> Fix IllegalArgumentException when down-converting from V2 to V0/V1
> KAFKA-6728  -
> Fix NullPointerException when instantiating the HeaderConverter
>
> Kafka 1.1.1 release plan:
> https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+1.1.1
>
> Release notes for the 1.1.1 release:
> http://home.apache.org/~lindong/kafka-1.1.1-rc3/RELEASE_NOTES.html
>
> *** Please download, test and vote by Thursday, July 12, 12pm PT ***
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~lindong/kafka-1.1.1-rc3/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~lindong/kafka-1.1.1-rc3/javadoc/
>
> * Tag to be voted upon (off 1.1 branch) is the 1.1.1-rc3 tag:
> https://github.com/apache/kafka/tree/1.1.1-rc3
>
> * Documentation:
> http://kafka.apache.org/11/documentation.html
>
> * Protocol:
> http://kafka.apache.org/11/protocol.html
>
> * Successful Jenkins builds for the 1.1 branch:
> Unit/integration tests: *https://builds.apache.org/job/kafka-1.1-jdk7/162
> *
> System tests:
> https://jenkins.confluent.io/job/system-test-kafka/job/1.1/156/
>
> Please test and verify the release artifacts and submit a vote for this RC,
> or report any issues so we can fix them and get a new RC out ASAP. Although
> this release vote requires PMC votes to pass, testing, votes, and bug
> reports are valuable and appreciated from everyone.
>
>
> Regards,
> Dong
>

Re: Cannot run the inbuild zookeeper server.

[Ted Yu]

'Address already in use' indicates that there was some (zookeeper) server
running on the same machine.

Can you double check ?

Thanks

On Sun, Jul 1, 2018 at 4:45 AM, Vineet Herenj  wrote:

> Hello,
>
> I am not able to run the inbuild zookeeper server.
> Following error shows:
>
> *java.net.BindException: Address already in use*
>
> Please assist for the issue.
>
> Regards,
> *Vineet Apoorv Herenj*
> M.Tech, IIT Kharagpur
> Department of Computer Science and Engineering
> Kharagpur, India
>
> m: +919679700816
> e: *vinither...@gmail.com  *
>

Re: Partitions reassignment is failing in Kafka 1.1.0

[Ted Yu]

See KAFKA-7124

Cheers

On Sun, Jul 1, 2018 at 3:26 AM, Debraj Manna 
wrote:

> Is there any bug filed for this ?
>
> On Sun, Jul 1, 2018 at 6:37 AM, Ted Yu  wrote:
>
> > Code snippet from ReassignPartitionsCommand.scala :
> >
> >   "log_dirs" -> replicas.map(r =>
> > replicaLogDirAssignment.getOrElse(new TopicPartitionReplica(tp.topic,
> > tp.partition, r), AnyLogDir)).asJava
> >
> > We know that the appearance of "any" was due to the OrElse clause.
> > Arguably there is a bug in the above code that the number of AnyLogDir
> > should match the length of the replicas list, or "log_dirs" should be
> > omitted in such case.
> >
> > On Sat, Jun 30, 2018 at 12:06 AM, Manikumar 
> > wrote:
> >
> > > It will be taken as "any" directory for each replica, which means
> replica
> > > will placed on any one of the
> > > configured directory on that broker.
> > >
> > > Since it is "log_dirs" optional, you can remove from the json.
> > >
> > > On Sat, Jun 30, 2018 at 12:02 PM Debraj Manna <
> subharaj.ma...@gmail.com>
> > > wrote:
> > >
> > > > It is problem on my side. The code was changing the replicas count
> but
> > > not
> > > > the log_dirs. Since I am migrating from 0.10 this part of the code
> was
> > > not
> > > > changed.
> > > >
> > > > I have a follow up question what is the default value of log_dirs if
> I
> > > > don't specify it in reassignment.json ?
> > > >
> > > > On Sat, Jun 30, 2018 at 11:15 AM, Debraj Manna <
> > subharaj.ma...@gmail.com
> > > >
> > > > wrote:
> > > >
> > > > > I am generating the reassignent.json like below
> > > > >
> > > > > /home/ubuntu/deploy/kafka/bin/kafka-reassign-partitions.sh
> > --zookeeper
> > > > 127.0.0.1:2181 --generate --topics-to-move-json-file
> > > > /home/ubuntu/deploy/kafka/topics_to_move.json --broker-list '%s'
> |tail
> > > -1 >
> > > > /home/ubuntu/deploy/kafka/reassignment.json"
> > > > >
> > > > > Then I am doing the reassignment using the generated file
> > > > >
> > > > > /home/ubuntu/deploy/kafka/bin/kafka-reassign-partitions.sh
> > --zookeeper
> > > > 127.0.0.1:2181 --execute --reassignment-json-file
> > > > /home/ubuntu/deploy/kafka/reassignment.json
> > > > >
> > > > > kafka-reassign-partitions.sh helps states
> > > > >
> > > > > The JSON file with the partition reassignment configurationThe
> format
> > > to
> > > > >> use is -
> > > > >> {"partitions":[{"topic": "foo", "partition": 1, "replicas":
> [1,2,3],
> > > > >> "log_dirs": ["dir1","dir2","dir3"]}], "version":1} Note that
> > > "log_dirs"
> > > > is
> > > > >> optional. When it is specified, its length must equal the length
> of
> > > the
> > > > >> replicas list. The value in this list can be either "any" or the
> > > > absolution
> > > > >> path of the log directory on the broker. If absolute log directory
> > > path
> > > > is
> > > > >> specified, it is currently required that the replica has not
> already
> > > > been
> > > > >> created on that broker. The replica will then be created in the
> > > > specified
> > > > >> log directory on the broker later.
> > > > >
> > > > >
> > > > > So it appears reassignment json that is generated by
> > > > > kafka-reassign-partions.sh is creating an issue with logdirs. Is
> this
> > > > > some issue in kafka-reassign-partitions.sh or some misconfiguration
> > > from
> > > > my
> > > > > side. ?
> > > > >
> > > > > On Sat, Jun 30, 2018 at 10:26 AM, Debraj Manna <
> > > subharaj.ma...@gmail.com
> > > > >
> > > > > wrote:
> > > > >
> > > > >> Please find the server.properties from one of the broker.
> > > > >>
> > > > >> broker.id=0
> > > > >> port=9092
> > > > >> num.network.threads=3
> > > > >> n

Re: Partitions reassignment is failing in Kafka 1.1.0

[Ted Yu]

Code snippet from ReassignPartitionsCommand.scala :

  "log_dirs" -> replicas.map(r =>
replicaLogDirAssignment.getOrElse(new TopicPartitionReplica(tp.topic,
tp.partition, r), AnyLogDir)).asJava

We know that the appearance of "any" was due to the OrElse clause.
Arguably there is a bug in the above code that the number of AnyLogDir
should match the length of the replicas list, or "log_dirs" should be
omitted in such case.

On Sat, Jun 30, 2018 at 12:06 AM, Manikumar 
wrote:

> It will be taken as "any" directory for each replica, which means replica
> will placed on any one of the
> configured directory on that broker.
>
> Since it is "log_dirs" optional, you can remove from the json.
>
> On Sat, Jun 30, 2018 at 12:02 PM Debraj Manna 
> wrote:
>
> > It is problem on my side. The code was changing the replicas count but
> not
> > the log_dirs. Since I am migrating from 0.10 this part of the code was
> not
> > changed.
> >
> > I have a follow up question what is the default value of log_dirs if I
> > don't specify it in reassignment.json ?
> >
> > On Sat, Jun 30, 2018 at 11:15 AM, Debraj Manna  >
> > wrote:
> >
> > > I am generating the reassignent.json like below
> > >
> > > /home/ubuntu/deploy/kafka/bin/kafka-reassign-partitions.sh --zookeeper
> > 127.0.0.1:2181 --generate --topics-to-move-json-file
> > /home/ubuntu/deploy/kafka/topics_to_move.json --broker-list '%s' |tail
> -1 >
> > /home/ubuntu/deploy/kafka/reassignment.json"
> > >
> > > Then I am doing the reassignment using the generated file
> > >
> > > /home/ubuntu/deploy/kafka/bin/kafka-reassign-partitions.sh --zookeeper
> > 127.0.0.1:2181 --execute --reassignment-json-file
> > /home/ubuntu/deploy/kafka/reassignment.json
> > >
> > > kafka-reassign-partitions.sh helps states
> > >
> > > The JSON file with the partition reassignment configurationThe format
> to
> > >> use is -
> > >> {"partitions":[{"topic": "foo", "partition": 1, "replicas": [1,2,3],
> > >> "log_dirs": ["dir1","dir2","dir3"]}], "version":1} Note that
> "log_dirs"
> > is
> > >> optional. When it is specified, its length must equal the length of
> the
> > >> replicas list. The value in this list can be either "any" or the
> > absolution
> > >> path of the log directory on the broker. If absolute log directory
> path
> > is
> > >> specified, it is currently required that the replica has not already
> > been
> > >> created on that broker. The replica will then be created in the
> > specified
> > >> log directory on the broker later.
> > >
> > >
> > > So it appears reassignment json that is generated by
> > > kafka-reassign-partions.sh is creating an issue with logdirs. Is this
> > > some issue in kafka-reassign-partitions.sh or some misconfiguration
> from
> > my
> > > side. ?
> > >
> > > On Sat, Jun 30, 2018 at 10:26 AM, Debraj Manna <
> subharaj.ma...@gmail.com
> > >
> > > wrote:
> > >
> > >> Please find the server.properties from one of the broker.
> > >>
> > >> broker.id=0
> > >> port=9092
> > >> num.network.threads=3
> > >> num.io.threads=8
> > >> socket.send.buffer.bytes=102400
> > >> socket.receive.buffer.bytes=102400
> > >> socket.request.max.bytes=104857600
> > >> log.dirs=/var/lib/kafka/kafka-logs
> > >> num.recovery.threads.per.data.dir=1
> > >> log.retention.hours=36
> > >> log.retention.bytes=1073741824
> > >> log.segment.bytes=536870912
> > >> log.retention.check.interval.ms=30
> > >> log.cleaner.enable=false
> > >> zookeeper.connect=platform1:2181,platform2:2181,platform3:2181
> > >> message.max.bytes=1500
> > >> replica.fetch.max.bytes=1500
> > >> auto.create.topics.enable=true
> > >> zookeeper.connection.timeout.ms=6000
> > >> unclean.leader.election.enable=false
> > >> delete.topic.enable=false
> > >> offsets.topic.replication.factor=1
> > >> transaction.state.log.replication.factor=1
> > >> transaction.state.log.min.isr=1
> > >>
> > >> I have placed server.log from a broker at https://gist.github.com/deb
> > >> raj-

Re: [VOTE] 1.0.2 RC1

[Ted Yu]

+1

Ran test suite.

On Fri, Jun 29, 2018 at 10:02 PM, Matthias J. Sax 
wrote:

> Hello Kafka users, developers and client-developers,
>
> This is the second candidate for release of Apache Kafka 1.0.2.
>
> This is a bug fix release addressing 27 tickets:
> https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+1.0.2
>
> Release notes for the 1.0.2 release:
> http://home.apache.org/~mjsax/kafka-1.0.2-rc1/RELEASE_NOTES.html
>
> *** Please download, test and vote by end of next week (7/6/18).
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~mjsax/kafka-1.0.2-rc1/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~mjsax/kafka-1.0.2-rc1/javadoc/
>
> * Tag to be voted upon (off 1.0 branch) is the 1.0.2 tag:
> https://github.com/apache/kafka/releases/tag/1.0.2-rc1
>
> * Documentation:
> http://kafka.apache.org/10/documentation.html
>
> * Protocol:
> http://kafka.apache.org/10/protocol.html
>
> * Successful Jenkins builds for the 1.0 branch:
> Unit/integration tests: https://builds.apache.org/job/kafka-1.0-jdk7/214/
> System tests:
> https://jenkins.confluent.io/job/system-test-kafka/job/1.0/225/
>
> /**
>
> Thanks,
>   -Matthias
>
>
>

Re: Partitions reassignment is failing in Kafka 1.1.0

[Ted Yu]

Seems to be related to KIP-113.

server.properties didn't go thru. Do you mind pastebin'ing its content ?

If you can pastebin logs from broker, that should help.

Thanks

On Fri, Jun 29, 2018 at 10:37 AM, Debraj Manna 
wrote:

> Hi
>
> I altered a topic like below in kafka 1.1.0
>
> /home/ubuntu/deploy/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181
> --alter --topic Topic3 --config min.insync.replicas=2
>
> But whenever I am trying to verify the reassignment it is showing the
> below exception
>
> /home/ubuntu/deploy/kafka/bin/kafka-reassign-partitions.sh --zookeeper 
> 127.0.0.1:2181 --reassignment-json-file 
> /home/ubuntu/deploy/kafka/reassignment.json --verify
>
> Partitions reassignment failed due to Size of replicas list Vector(3, 0, 2) 
> is different from size of log dirs list Vector(any) for partition Topic3-7
> kafka.common.AdminCommandFailedException: Size of replicas list Vector(3, 0, 
> 2) is different from size of log dirs list Vector(any) for partition Topic3-7
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1$$anonfun$apply$4$$anonfun$apply$5.apply(ReassignPartitionsCommand.scala:262)
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1$$anonfun$apply$4$$anonfun$apply$5.apply(ReassignPartitionsCommand.scala:251)
>   at scala.collection.Iterator$class.foreach(Iterator.scala:891)
>   at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1$$anonfun$apply$4.apply(ReassignPartitionsCommand.scala:251)
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1$$anonfun$apply$4.apply(ReassignPartitionsCommand.scala:250)
>   at scala.collection.immutable.List.foreach(List.scala:392)
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1.apply(ReassignPartitionsCommand.scala:250)
>   at 
> kafka.admin.ReassignPartitionsCommand$$anonfun$parsePartitionReassignmentData$1.apply(ReassignPartitionsCommand.scala:249)
>   at scala.collection.immutable.List.foreach(List.scala:392)
>   at 
> kafka.admin.ReassignPartitionsCommand$.parsePartitionReassignmentData(ReassignPartitionsCommand.scala:249)
>   at 
> kafka.admin.ReassignPartitionsCommand$.verifyAssignment(ReassignPartitionsCommand.scala:90)
>   at 
> kafka.admin.ReassignPartitionsCommand$.verifyAssignment(ReassignPartitionsCommand.scala:84)
>   at 
> kafka.admin.ReassignPartitionsCommand$.main(ReassignPartitionsCommand.scala:58)
>   at 
> kafka.admin.ReassignPartitionsCommand.main(ReassignPartitionsCommand.scala)
>
>
> My reassignment.json & server.properties is attached. Same thing used to
> work fine in kafka 0.10. Can someone let me what is going wrong? Is
> anything changed related to this in kafka 1.1.0 ?
>

Re: [VOTE] 0.10.2.2 RC1

[Ted Yu]

+1

Ran test suite.

Checked signatures.

On Fri, Jun 29, 2018 at 10:21 AM, Jason Gustafson 
wrote:

> +1 (binding). I checked release notes, documentation, and went through the
> quickstart.
>
> Thanks Matthias!
>
> On Fri, Jun 22, 2018 at 6:43 PM, Matthias J. Sax 
> wrote:
>
> > Hello Kafka users, developers and client-developers,
> >
> > This is the second candidate for release of Apache Kafka 0.10.2.2.
> >
> > Note, that RC0 was created before the upgrade to Gradle 4.8.1 and thus,
> > we discarded it in favor of RC1 (without sending out a email for RC0).
> >
> > This is a bug fix release closing 29 tickets:
> > https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.10.2.2
> >
> > Release notes for the 0.10.2.2 release:
> > http://home.apache.org/~mjsax/kafka-0.10.2.2-rc1/RELEASE_NOTES.html
> >
> > *** Please download, test and vote by Tuesday, 6/26/18 end-of-day, so we
> > can close the vote on Wednesday.
> >
> > Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > * Release artifacts to be voted upon (source and binary):
> > http://home.apache.org/~mjsax/kafka-0.10.2.2-rc1/
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Javadoc:
> > http://home.apache.org/~mjsax/kafka-0.10.2.2-rc1/javadoc/
> >
> > * Tag to be voted upon (off 0.10.2 branch) is the 0.10.2.2 tag:
> > https://github.com/apache/kafka/releases/tag/0.10.2.2-rc1
> >
> > * Documentation:
> > http://kafka.apache.org/0102/documentation.html
> >
> > * Protocol:
> > http://kafka.apache.org/0102/protocol.html
> >
> > * Successful Jenkins builds for the 0.10.2 branch:
> > Unit/integration tests: https://builds.apache.org/job/
> > kafka-0.10.2-jdk7/220/
> >
> > /**
> >
> > Thanks,
> >   -Matthias
> >
> >
>

Re: [DISCUSS] KIP-328: Ability to suppress updates for KTables

[Ted Yu]

I noticed this (lack of primary parameter) as well.

What you gave as new example is semantically the same as what I suggested.
So it is good by me.

Thanks

On Wed, Jun 27, 2018 at 7:31 AM, John Roesler  wrote:

> Thanks for taking look, Ted,
>
> I agree this is a departure from the conventions of Streams DSL.
>
> Most of our config objects have one or two "required" parameters, which fit
> naturally with the static factory method approach. TimeWindow, for example,
> requires a size parameter, so we can naturally say TimeWindows.of(size).
>
> I think in the case of a suppression, there's really no "core" parameter,
> and "Suppression.of()" seems sillier than "new Suppression()". I think that
> Suppression.of(duration) would be ambiguous, since there are many durations
> that we can configure.
>
> However, thinking about it again, I suppose that I can give each
> configuration method a static version, which would let you replace "new
> Suppression()." with "Suppression." in all the examples. Basically, instead
> of "of()", we'd support any of the methods I listed.
>
> For example:
>
> windowCounts
> .suppress(
> Suppression
> .suppressLateEvents(Duration.ofMinutes(10))
> .suppressIntermediateEvents(
> IntermediateSuppression.emitAfter(Duration.ofMinutes(10))
> )
> );
>
>
> Does that seem better?
>
> Thanks,
> -John
>
>
> On Wed, Jun 27, 2018 at 12:44 AM Ted Yu  wrote:
>
> > I started to read this KIP which contains a lot of materials.
> >
> > One suggestion:
> >
> > .suppress(
> > new Suppression()
> >
> >
> > Do you think it would be more consistent with the rest of Streams data
> > structures by supporting `of` ?
> >
> > Suppression.of(Duration.ofMinutes(10))
> >
> >
> > Cheers
> >
> >
> >
> > On Tue, Jun 26, 2018 at 1:11 PM, John Roesler  wrote:
> >
> > > Hello devs and users,
> > >
> > > Please take some time to consider this proposal for Kafka Streams:
> > >
> > > KIP-328: Ability to suppress updates for KTables
> > >
> > > link: https://cwiki.apache.org/confluence/x/sQU0BQ
> > >
> > > The basic idea is to provide:
> > > * more usable control over update rate (vs the current state store
> > caches)
> > > * the final-result-for-windowed-computations feature which several
> people
> > > have requested
> > >
> > > I look forward to your feedback!
> > >
> > > Thanks,
> > > -John
> > >
> >
>

Re: [VOTE] 0.11.0.3 RC0

[Ted Yu]

+1

Checked signatures.

Ran unit test suite.

On Fri, Jun 22, 2018 at 4:56 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> +1 (non-binding)
>
> Built from source and ran quickstart successfully on Ubuntu (with Java 8).
>
> Thanks Matthias!
> --Vahid
>
>
>
>
> From:   "Matthias J. Sax" 
> To: d...@kafka.apache.org, users@kafka.apache.org,
> kafka-clie...@googlegroups.com
> Date:   06/22/2018 03:14 PM
> Subject:[VOTE] 0.11.0.3 RC0
>
>
>
> Hello Kafka users, developers and client-developers,
>
> This is the first candidate for release of Apache Kafka 0.11.0.3.
>
> This is a bug fix release closing 27 tickets:
> https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.11.0.3
>
> Release notes for the 0.11.0.3 release:
> http://home.apache.org/~mjsax/kafka-0.11.0.3-rc0/RELEASE_NOTES.html
>
> *** Please download, test and vote by Tuesday, 6/26/18 end-of-day, so we
> can close the vote on Wednesday.
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~mjsax/kafka-0.11.0.3-rc0/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~mjsax/kafka-0.11.0.3-rc0/javadoc/
>
> * Tag to be voted upon (off 0.11.0 branch) is the 0.11.0.3 tag:
> https://github.com/apache/kafka/releases/tag/0.11.0.3-rc0
>
> * Documentation:
> http://kafka.apache.org/0110/documentation.html
>
> * Protocol:
> http://kafka.apache.org/0110/protocol.html
>
> * Successful Jenkins builds for the 0.11.0 branch:
> Unit/integration tests:
> https://builds.apache.org/job/kafka-0.11.0-jdk7/385/
> System tests:
> https://jenkins.confluent.io/job/system-test-kafka/job/0.11.0/217/
>
> /**
>
> Thanks,
>   -Matthias
>
> [attachment "signature.asc" deleted by Vahid S Hashemian/Silicon
> Valley/IBM]
>
>
>
>

Re: [VOTE] 1.1.1 RC1

[Ted Yu]

+1

Ran test suite.
Checked signatures.

On Fri, Jun 22, 2018 at 12:14 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> +1 (non-binding)
>
> Built from source and ran quickstart successfully on Ubuntu (with Java 8).
>
> Thanks Dong!
> --Vahid
>
>
>
> From:   Dong Lin 
> To: d...@kafka.apache.org, users@kafka.apache.org,
> kafka-clie...@googlegroups.com
> Date:   06/22/2018 10:10 AM
> Subject:[VOTE] 1.1.1 RC1
>
>
>
> Hello Kafka users, developers and client-developers,
>
> This is the second candidate for release of Apache Kafka 1.1.1.
>
> Apache Kafka 1.1.1 is a bug-fix release for the 1.1 branch that was first
> released with 1.1.0 about 3 months ago. We have fixed about 25 issues
> since
> that release. A few of the more significant fixes include:
>
> KAFKA-6925 <
> https://issues.apache.org/jira/browse/KAFKA-6925
> > - Fix memory
> leak in StreamsMetricsThreadImpl
> KAFKA-6937 <
> https://issues.apache.org/jira/browse/KAFKA-6937
> > - In-sync
> replica delayed during fetch if replica throttle is exceeded
> KAFKA-6917 <
> https://issues.apache.org/jira/browse/KAFKA-6917
> > - Process txn
> completion asynchronously to avoid deadlock
> KAFKA-6893 <
> https://issues.apache.org/jira/browse/KAFKA-6893
> > - Create
> processors before starting acceptor to avoid ArithmeticException
> KAFKA-6870 <
> https://issues.apache.org/jira/browse/KAFKA-6870
> > -
> Fix ConcurrentModificationException in SampledStat
> KAFKA-6878 <
> https://issues.apache.org/jira/browse/KAFKA-6878
> > - Fix
> NullPointerException when querying global state store
> KAFKA-6879 <
> https://issues.apache.org/jira/browse/KAFKA-6879
> > - Invoke
> session init callbacks outside lock to avoid Controller deadlock
> KAFKA-6857 <
> https://issues.apache.org/jira/browse/KAFKA-6857
> > - Prevent
> follower from truncating to the wrong offset if undefined leader epoch is
> requested
> KAFKA-6854 <
> https://issues.apache.org/jira/browse/KAFKA-6854
> > - Log cleaner
> fails with transaction markers that are deleted during clean
> KAFKA-6747 <
> https://issues.apache.org/jira/browse/KAFKA-6747
> > - Check
> whether there is in-flight transaction before aborting transaction
> KAFKA-6748 <
> https://issues.apache.org/jira/browse/KAFKA-6748
> > - Double
> check before scheduling a new task after the punctuate call
> KAFKA-6739 <
> https://issues.apache.org/jira/browse/KAFKA-6739
> > -
> Fix IllegalArgumentException when down-converting from V2 to V0/V1
> KAFKA-6728 <
> https://issues.apache.org/jira/browse/KAFKA-6728
> > -
> Fix NullPointerException when instantiating the HeaderConverter
>
> Kafka 1.1.1 release plan:
> https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+1.1.1
>
>
> Release notes for the 1.1.1 release:
> http://home.apache.org/~lindong/kafka-1.1.1-rc1/RELEASE_NOTES.html
>
>
> *** Please download, test and vote by Thursday, Jun 22, 12pm PT ***
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~lindong/kafka-1.1.1-rc1/
>
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
>
> * Javadoc:
> http://home.apache.org/~lindong/kafka-1.1.1-rc1/javadoc/
>
>
> * Tag to be voted upon (off 1.1 branch) is the 1.1.1-rc1 tag:
> https://github.com/apache/kafka/tree/1.1.1-rc1
>
>
> * Documentation:
> http://kafka.apache.org/11/documentation.html
>
>
> * Protocol:
> http://kafka.apache.org/11/protocol.html
>
>
> * Successful Jenkins builds for the 1.1 branch:
> Unit/integration tests:
> *https://builds.apache.org/job/kafka-1.1-jdk7/152/
> <
> https://builds.apache.org/job/kafka-1.1-jdk7/152/
> >*
> System tests:
> https://jenkins.confluent.io/job/system-test-
>
> kafka-branch-builder/1817
>
>
> Please test and verify the release artifacts and submit a vote for this
> RC,
> or report any issues so we can fix them and get a new RC out ASAP.
> Although
> this release vote requires PMC votes to pass, testing, votes, and bug
> reports are valuable and appreciated from everyone.
>
> Cheers,
> Dong
>
>
>
>
>

Re: [VOTE] 1.1.1 RC0

[Ted Yu]

+1

Ran unit test suite which passed.

Checked signatures.

On Tue, Jun 19, 2018 at 4:47 PM, Dong Lin  wrote:

> Re-send to kafka-clie...@googlegroups.com
>
> On Tue, Jun 19, 2018 at 4:29 PM, Dong Lin  wrote:
>
> > Hello Kafka users, developers and client-developers,
> >
> > This is the first candidate for release of Apache Kafka 1.1.1.
> >
> > Apache Kafka 1.1.1 is a bug-fix release for the 1.1 branch that was first
> > released with 1.1.0 about 3 months ago. We have fixed about 25 issues
> since
> > that release. A few of the more significant fixes include:
> >
> > KAFKA-6925  - Fix
> > memory leak in StreamsMetricsThreadImpl
> > KAFKA-6937  - In-sync
> > replica delayed during fetch if replica throttle is exceeded
> > KAFKA-6917  - Process
> > txn completion asynchronously to avoid deadlock
> > KAFKA-6893  - Create
> > processors before starting acceptor to avoid ArithmeticException
> > KAFKA-6870  -
> > Fix ConcurrentModificationException in SampledStat
> > KAFKA-6878  - Fix
> > NullPointerException when querying global state store
> > KAFKA-6879  - Invoke
> > session init callbacks outside lock to avoid Controller deadlock
> > KAFKA-6857  - Prevent
> > follower from truncating to the wrong offset if undefined leader epoch is
> > requested
> > KAFKA-6854  - Log
> > cleaner fails with transaction markers that are deleted during clean
> > KAFKA-6747  - Check
> > whether there is in-flight transaction before aborting transaction
> > KAFKA-6748  - Double
> > check before scheduling a new task after the punctuate call
> > KAFKA-6739  -
> > Fix IllegalArgumentException when down-converting from V2 to V0/V1
> > KAFKA-6728  -
> > Fix NullPointerException when instantiating the HeaderConverter
> >
> > Kafka 1.1.1 release plan:
> > https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+1.1.1
> >
> > Release notes for the 1.1.1 release:
> > http://home.apache.org/~lindong/kafka-1.1.1-rc0/RELEASE_NOTES.html
> >
> > *** Please download, test and vote by Thursday, Jun 22, 12pm PT ***
> >
> > Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > * Release artifacts to be voted upon (source and binary):
> > http://home.apache.org/~lindong/kafka-1.1.1-rc0/
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Tag to be voted upon (off 1.1 branch) is the 1.1.1-rc0 tag:
> > https://github.com/apache/kafka/tree/1.1.1-rc0
> >
> > * Documentation:
> > http://kafka.apache.org/11/documentation.html
> >
> > * Protocol:
> > http://kafka.apache.org/11/protocol.html
> >
> > * Successful Jenkins builds for the 1.1 branch:
> > Unit/integration tests: https://builds.apache.org/job/
> kafka-1.1-jdk7/150/
> >
> > Please test and verify the release artifacts and submit a vote for this
> RC,
> > or report any issues so we can fix them and get a new RC out ASAP.
> Although
> > this release vote requires PMC votes to pass, testing, votes, and bug
> > reports are valuable and appreciated from everyone.
> >
> > Cheers,
> > Dong
> >
> >
> >
>

Re: NPE in low level Kafka topology

[Ted Yu]

Can you show the related code from OneToManyGroupedProcessor ?

Thanks

On Mon, Jun 18, 2018 at 4:29 AM, Frank Lyaruu  wrote:

> Hi, I've upgraded our 0.11 based stream application to the trunk version,
> and I get an intermittent NPE. It's is quite a big topology, and I haven't
> succeeded in reproducing it on a simpler topology.
> It builds the topology, starts Kafka Streams, runs for about 20s., and then
> it terminates
> It seems that the 'currentNode' in the ProcessorContext is null.
>
> Does this ring a bell for anyone?
>
> [lowlevel-develop-generation-7aa-lowlevel-e23137ae-d94b-
> 4f17-a684-995320fd426d-StreamThread-12]
> ERROR org.apache.kafka.streams.processor.internals.AssignedStreamsTasks -
> stream-thread
> [lowlevel-develop-generation-7aa-lowlevel-e23137ae-d94b-
> 4f17-a684-995320fd426d-StreamThread-12]
> Failed to process stream task 0_0 due to the following error:
> java.lang.NullPointerException
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:114)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:90)
> at
> com.dexels.kafka.streams.remotejoin.ranged.OneToManyGroupedProcessor.
> forwardMessage(OneToManyGroupedProcessor.java:125)
> at
> com.dexels.kafka.streams.remotejoin.ranged.OneToManyGroupedProcessor.
> forwardJoin(OneToManyGroupedProcessor.java:101)
> at
> com.dexels.kafka.streams.remotejoin.ranged.OneToManyGroupedProcessor.
> process(OneToManyGroupedProcessor.java:70)
> at
> com.dexels.kafka.streams.remotejoin.ranged.OneToManyGroupedProcessor.
> process(OneToManyGroupedProcessor.java:1)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(
> ProcessorNode.java:50)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.
> runAndMeasureLatency(ProcessorNode.java:244)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.process(
> ProcessorNode.java:133)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:143)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:126)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:90)
> at
> com.dexels.kafka.streams.remotejoin.PreJoinProcessor.
> process(PreJoinProcessor.java:25)
> at
> com.dexels.kafka.streams.remotejoin.PreJoinProcessor.
> process(PreJoinProcessor.java:1)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(
> ProcessorNode.java:50)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.
> runAndMeasureLatency(ProcessorNode.java:244)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.process(
> ProcessorNode.java:133)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:143)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:126)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:90)
> at
> com.dexels.kafka.streams.remotejoin.StoreProcessor.
> process(StoreProcessor.java:48)
> at
> com.dexels.kafka.streams.remotejoin.StoreProcessor.
> process(StoreProcessor.java:1)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(
> ProcessorNode.java:50)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.
> runAndMeasureLatency(ProcessorNode.java:244)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.process(
> ProcessorNode.java:133)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:143)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:126)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:90)
> at
> com.dexels.kafka.streams.remotejoin.XmlTransformerProcessor.process(
> XmlTransformerProcessor.java:52)
> at
> com.dexels.kafka.streams.remotejoin.XmlTransformerProcessor.process(
> XmlTransformerProcessor.java:1)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(
> ProcessorNode.java:50)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.
> runAndMeasureLatency(ProcessorNode.java:244)
> at
> org.apache.kafka.streams.processor.internals.ProcessorNode.process(
> ProcessorNode.java:133)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:143)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> ProcessorContextImpl.java:126)
> at
> org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(
> 

Re: INVALID_FETCH_SESSION_EPOCH after upgrade to 1.1.0

[Ted Yu]

In log4j.properties, can you make the following change (you can keep
whatever follows the first comma in the rootLogger line):

log4j.rootLogger=DEBUG
log4j.logger.org.apache.kafka=DEBUG

FetchSession.scala is in kafka.server package. You can just turn on DEBUG
for this package.

FYI


On Wed, Jun 13, 2018 at 9:08 AM, Mark Anderson 
wrote:

> Ted
>
> I don't see any other INFO log messages so I assume that means it is the
> DEBUG case I'm seeing?
>
> I don't have DEBUG enabled at the moment.
>
> Thanks
>
> On Wed, 13 Jun 2018, 00:21 Ted Yu,  wrote:
>
> > Before Errors.INVALID_FETCH_SESSION_EPOCH is returned,
> FetchSession.scala
> > would log the reason for the response.
> > There are 3 cases, 2 with info log and 1 with debug log.
> > Here is one code snippet:
> >
> > if (session.epoch != reqMetadata.epoch()) {
> >
> >   debug(s"Created a new error FetchContext for session id ${
> > session.id}: expected " +
> >
> > s"epoch ${session.epoch}, but got epoch $
> > {reqMetadata.epoch()}.")
> >
> >   new SessionErrorContext(Errors.
> INVALID_FETCH_SESSION_EPOCH,
> > reqMetadata)
> >
> > Can you pastebin the log line preceding what you pasted ?
> >
> > Thanks
> >
> > On Tue, Jun 12, 2018 at 3:12 PM, Mark Anderson 
> > wrote:
> >
> > > We recently updated our Kafka brokers and clients to 1.1.0. Since the
> > > upgrade we periodically see INFO log entries such as
> > >
> > > INFO Jun 08 08:30:20.335 61161458 [KafkaRecordConsumer-0]
> > > org.apache.kafka.clients.FetchSessionHandler [Consumer
> > > clientId=consumer-1,
> > > groupId=group_60_10] Node 3 was unable to process the fetch request
> with
> > > (sessionId=819759315, epoch=145991): INVALID_FETCH_SESSION_EPOCH.
> > >
> > > I see that this message comes from the changes introduced in KIP-227:
> > > Introduce Incremental FetchRequests To Increase Partition Stability
> > > <https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > > 227%3A+Introduce+Incremental+FetchRequests+to+Increase+
> > > Partition+Scalability>.
> > > However, I cannot find any detailed information about why this message
> > > would appear or what parameters might have to be tuned after its
> > > introduction.
> > >
> > > So far it doesn't seem to have an impact on consumer behaviour with
> > respect
> > > to receiving records but I would like to understand
> > >
> > >1. Why is the message being logged?
> > >2. Do I need to do anything?
> > >3. Can anything be done to stop it being logged?
> > >
> > > Thanks,
> > > Mark
> > >
> >
>

Re: Details of segment deletion

[Ted Yu]

Minor clarification (since new segment appeared twice) :

bq. before a new one is deleted.

The 'new one' (in the last sentence) would become old when another segment
is created.

Cheers


On Tue, Jun 12, 2018 at 6:42 PM, Gwen Shapira  wrote:

> See below:
>
> On Mon, Jun 11, 2018 at 3:36 AM, Simon Cooper <
> simon.coo...@featurespace.co.uk> wrote:
>
> > Hi,
> >
> > I've ben trying to work out the details of when exactly kafka log
> segments
> > get deleted for to the retention period, so it would be helpful if
> someone
> > could clarify the behaviour:
> >
> >
> >   *   Is a segment only deleted when all messages in that segment have
> > 'timed out', or are messages deleted within each segment?
> >
>
> Kafka only deletes entire segments (except for compacted topics, which are
> a different story)
>
>
>
> >   *   Does the server artificially limit the messages returned to clients
> > to those within the retention period, even if they still exist in the
> > segment file?
> >
>
> Older messages can be read if the segment wasn't deleted yet. You can check
> the "beginning of log" offset JMX metric to see what is the oldest offset
> available to consumers on each partition.
>
>
> >   *   Does the segment deletion happen when a new segment is created, or
> > is it done as a separate operation by the log cleaner?
> >
>
> Separate operation by log cleaner, but note that active segment is never
> deleted so sometimes you are waiting for new segment to get created before
> a new one is deleted.
>
>
> >
> > Thanks for the help!
> > Simon Cooper
> >
>
>
>
> --
> *Gwen Shapira*
> Product Manager | Confluent
> 650.450.2760 | @gwenshap
> Follow us: Twitter  | blog
> 
>

Re: INVALID_FETCH_SESSION_EPOCH after upgrade to 1.1.0

[Ted Yu]

Before Errors.INVALID_FETCH_SESSION_EPOCH is returned, FetchSession.scala
would log the reason for the response.
There are 3 cases, 2 with info log and 1 with debug log.
Here is one code snippet:

if (session.epoch != reqMetadata.epoch()) {

  debug(s"Created a new error FetchContext for session id ${
session.id}: expected " +

s"epoch ${session.epoch}, but got epoch $
{reqMetadata.epoch()}.")

  new SessionErrorContext(Errors.INVALID_FETCH_SESSION_EPOCH,
reqMetadata)

Can you pastebin the log line preceding what you pasted ?

Thanks

On Tue, Jun 12, 2018 at 3:12 PM, Mark Anderson 
wrote:

> We recently updated our Kafka brokers and clients to 1.1.0. Since the
> upgrade we periodically see INFO log entries such as
>
> INFO Jun 08 08:30:20.335 61161458 [KafkaRecordConsumer-0]
> org.apache.kafka.clients.FetchSessionHandler [Consumer
> clientId=consumer-1,
> groupId=group_60_10] Node 3 was unable to process the fetch request with
> (sessionId=819759315, epoch=145991): INVALID_FETCH_SESSION_EPOCH.
>
> I see that this message comes from the changes introduced in KIP-227:
> Introduce Incremental FetchRequests To Increase Partition Stability
>  227%3A+Introduce+Incremental+FetchRequests+to+Increase+
> Partition+Scalability>.
> However, I cannot find any detailed information about why this message
> would appear or what parameters might have to be tuned after its
> introduction.
>
> So far it doesn't seem to have an impact on consumer behaviour with respect
> to receiving records but I would like to understand
>
>1. Why is the message being logged?
>2. Do I need to do anything?
>3. Can anything be done to stop it being logged?
>
> Thanks,
> Mark
>

Re: Kafka Scala debugging and diagnostics

[Ted Yu]

Can you share what you plan to write (on the mailing list) ?

Thanks

On Thu, May 17, 2018 at 9:15 AM, M. Manna <manme...@gmail.com> wrote:

> I have had some success.
>
> Is it okay for us to update Cwiki with the setup steps which I've got ? or
> too generic?
>
> On 16 May 2018 at 23:04, M. Manna <manme...@gmail.com> wrote:
>
> > Must have missed that in Kafka-run-class.bat which sets suspend=n by
> > default. Thanks for pointing that out, giving it a try.
> >
> >
> >
> > And didn
> >
> > On Wed, 16 May 2018 at 22:57, Ted Yu <yuzhih...@gmail.com> wrote:
> >
> >> Have you set the following two environment variables:
> >>
> >> export KAFKA_DEBUG=y; export DEBUG_SUSPEND_FLAG=y;
> >>
> >> The above would freeze the process until debugger is attached.
> >>
> >> Cheers
> >>
> >> On Wed, May 16, 2018 at 2:46 PM, M. Manna <manme...@gmail.com> wrote:
> >>
> >> > Hello,
> >> >
> >> > Is there anyone who can provide a few steps for setting up Eclipse IDE
> >> to
> >> > debug Kafka cluster ?  Has anyone got eclipse/scala IDE experience in
> >> Kafka
> >> > debugging ?
> >> >
> >> > I have been trying for days, but couldn’t start the debugger to
> connect
> >> to
> >> > port 5005 when the Cluster is running. Everything else is set up call
> as
> >> > per github and cwiki docs.
> >> >
> >> > Thanks in advance,
> >> >
> >>
> >
>

Re: Kafka Scala debugging and diagnostics

[Ted Yu]

Have you set the following two environment variables:

export KAFKA_DEBUG=y; export DEBUG_SUSPEND_FLAG=y;

The above would freeze the process until debugger is attached.

Cheers

On Wed, May 16, 2018 at 2:46 PM, M. Manna  wrote:

> Hello,
>
> Is there anyone who can provide a few steps for setting up Eclipse IDE to
> debug Kafka cluster ?  Has anyone got eclipse/scala IDE experience in Kafka
> debugging ?
>
> I have been trying for days, but couldn’t start the debugger to connect to
> port 5005 when the Cluster is running. Everything else is set up call as
> per github and cwiki docs.
>
> Thanks in advance,
>

Re: Kafka 2.11-1.1.0 crashes brokers and brings cluster down on Windows

[Ted Yu]

.WindowsException.translateToIOException(
> WindowsException.java:86)
> at sun.nio.fs.WindowsException.rethrowAsIOException(
> WindowsException.java:97)
> at sun.nio.fs.WindowsException.rethrowAsIOException(
> WindowsException.java:102)
> at sun.nio.fs.WindowsFileSystemProvider.implDelete(
> WindowsFileSystemProvider.java:269)
> at sun.nio.fs.AbstractFileSystemProvider.deleteIfExists(
> AbstractFileSystemProvider.java:108)
> at java.nio.file.Files.deleteIfExists(Files.java:1165)
> at kafka.log.Cleaner.deleteCleanedFileIfExists$1(
> LogCleaner.scala:488)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:493)
> at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.
> scala:462)
> at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.
> scala:461)
> at scala.collection.immutable.List.foreach(List.scala:392)
> at kafka.log.Cleaner.doClean(LogCleaner.scala:461)
> at kafka.log.Cleaner.clean(LogCleaner.scala:438)
> at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(
> LogCleaner.scala:305)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:291)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
> [2018-05-14 16:35:29,345] INFO [ReplicaFetcherManager on broker 1] Removed
> fetcher for partitions  (kafka.server.ReplicaFetcherManager)
> [2018-05-14 16:35:29,361] INFO Registered broker 1 at path /brokers/ids/1
> with addresses: 
> ArrayBuffer(EndPoint(localhost,9092,ListenerName(PLAINTEXT),PLAINTEXT))
> (kafka.zk.KafkaZkClient)
> [2018-05-14 16:35:29,392] INFO [ReplicaAlterLogDirsManager on broker 1]
> Removed fetcher for partitions  (kafka.server.ReplicaAlterLogDirsManager)
> [2018-05-14 16:35:29,424] INFO [ReplicaManager broker=1] Broker 1 stopped
> fetcher for partitions  and stopped moving logs for partitions  because
> they are in the failed log directory C:\kafka1.
> (kafka.server.ReplicaManager)
> [2018-05-14 16:35:29,424] ERROR Failed to clean up log for
> __consumer_offsets-21 in dir C:\kafka1 due to IOException (kafka.server.
> LogDirFailureChannel)
> java.nio.file.FileSystemException: C:\kafka1\__consumer_offsets-
> 21\.log.cleaned: The process cannot access the file
> because it is being used by another process.
>
> at sun.nio.fs.WindowsException.translateToIOException(
> WindowsException.java:86)
> at sun.nio.fs.WindowsException.rethrowAsIOException(
> WindowsException.java:97)
> at sun.nio.fs.WindowsException.rethrowAsIOException(
> WindowsException.java:102)
> at sun.nio.fs.WindowsFileSystemProvider.implDelete(
> WindowsFileSystemProvider.java:269)
> at sun.nio.fs.AbstractFileSystemProvider.deleteIfExists(
> AbstractFileSystemProvider.java:108)
> at java.nio.file.Files.deleteIfExists(Files.java:1165)
> at kafka.log.Cleaner.deleteCleanedFileIfExists$1(
> LogCleaner.scala:488)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:493)
> at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.
> scala:462)
> at kafka.log.Cleaner$$anonfun$doClean$4.apply(LogCleaner.
> scala:461)
> at scala.collection.immutable.List.foreach(List.scala:392)
> at kafka.log.Cleaner.doClean(LogCleaner.scala:461)
> at kafka.log.Cleaner.clean(LogCleaner.scala:438)
> at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(
> LogCleaner.scala:305)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:291)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
> [2018-05-14 16:35:29,439] INFO Stopping serving logs in dir C:\kafka1
> (kafka.log.LogManager)
> [2018-05-14 16:35:29,517] INFO [ExpirationReaper-1-topic]: Starting
> (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
> [2018-05-14 16:35:29,517] ERROR Shutdown broker because all log dirs in
> C:\kafka1 have failed (kafka.log.LogManager)
>
> Attached is the full log upon kafka-server-start.bat run. Meanwhile, i am
> experimenting whether closing a log and txnIndex (if not closed already)
> and unmapping the other segments before renaming helps. The problem is that
> if .append() is having a clash then this approach might not help
> (Potentially ClosedChannelException if the fileChannel is being closed
> twice).
> Once again, thanks a lot for helping out on this.
>
> Regards,
>
>
> On 14 May 2018 at 16:55, Ted Yu <yuzhih...@gmail.com> wrote:
>
>> Looks like you were trying to show some screen which didn't go thru.
>>
>> Examining LogSegment.scala, I wonder if the time index was being appended
>> to (which caused the deletion to fail):
>>
>>   // append an entry to the index (if neede

Re: Kafka 2.11-1.1.0 crashes brokers and brings cluster down on Windows

[Ted Yu]

Looks like you were trying to show some screen which didn't go thru.

Examining LogSegment.scala, I wonder if the time index was being appended
to (which caused the deletion to fail):

  // append an entry to the index (if needed)
  if(bytesSinceLastIndexEntry > indexIntervalBytes) {
offsetIndex.append(largestOffset, physicalPosition)
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestamp)

Can you capture the stack trace and pastebin it ?

If LogSegment shows up in the stack trace, we may have some idea for the
root cause.

Thanks

On Mon, May 14, 2018 at 8:31 AM, M. Manna <manme...@gmail.com> wrote:

> I am having some difficulties debugging the cluster. IN CWIKI for kafka,
> I can see a setup guide for eclipse, but nothing as such for debugger
> setup. The issue is that every time the log cleaner thread is run, it's
> having a FATAL shutdown.
> I tried to close all producers and consumers, but it still locks the log
> and Transaction index segments. Forcefully unmapping the offset and time
> offset files work normally, but log and transaction index files don't work
> properly.
>
> Could anyone please advise how to setup the kafka debugging on eclipse ? I
> have added the sdb jar noted in scala-debugger.org site. But I couldn't
> connect to port 5005 at all.
>
>
>
>
> On 13 May 2018 at 10:00, M. Manna <manme...@gmail.com> wrote:
>
>> Hi Ted,
>>
>> I highly appreciate the response over the weekend, and thanks for
>> pointing out the JIRAs.
>>
>> I don't believe the processes are responsible, but individual threads
>> which are still holding the log/index files using IO streams. I am trying
>> walk a single node setup through debugger to find out which thread is
>> locking the file. Apologise but it's a huge application so it might get me
>> some time to get around it :)
>>
>> Please do update if you find something new.
>>
>> Regards,
>>
>> On 12 May 2018 at 22:15, Ted Yu <yuzhih...@gmail.com> wrote:
>>
>>> There are two outstanding issues: KAFKA-6059 and KAFKA-6200 which bear
>>> some
>>> resemblance.
>>>
>>> Can you try to find how the other process uses the file being deleted ?
>>>
>>> https://superuser.com/questions/117902/find-out-which-proces
>>> s-is-locking-a-file-or-folder-in-windows
>>> https://www.computerhope.com/issues/ch000714.htm
>>>
>>> Cheers
>>>
>>> On Sat, May 12, 2018 at 1:42 PM, M. Manna <manme...@gmail.com> wrote:
>>>
>>> > Hello,
>>> >
>>> > We are still stuck with this issue where 2.11-1.1.0 distro is failing
>>> to
>>> > cleanup logs on Windows and brings the entire cluster down one by one.
>>> > Extending the retention hours and sizes don't help because they burden
>>> the
>>> > hard drive.
>>> >
>>> > Here is the log
>>> >
>>> > [2018-05-12 21:36:57,673] INFO [Log partition=test-0, dir=C:\kafka1]
>>> Rolled
>>> > > new log segment at offset 45 in 105 ms. (kafka.log.Log)
>>> > > [2018-05-12 21:36:57,673] INFO [Log partition=test-0, dir=C:\kafka1]
>>> > > Scheduling log segment [baseOffset 0, size 2290] for deletion.
>>> > > (kafka.log.Log)
>>> > > [2018-05-12 21:36:57,673] ERROR Error while deleting segments for
>>> test-0
>>> > > in dir C:\kafka1 (kafka.server.LogDirFailureChannel)
>>> > > java.nio.file.FileSystemException:
>>> > > C:\kafka1\test-0\.log ->
>>> > > C:\kafka1\test-0\.log.deleted: The process
>>> cannot
>>> > > access the file because it is being used by another process.
>>> > >
>>> > > at
>>> > > sun.nio.fs.WindowsException.translateToIOException(
>>> > WindowsException.java:86)
>>> > > at
>>> > > sun.nio.fs.WindowsException.rethrowAsIOException(
>>> > WindowsException.java:97)
>>> > > at sun.nio.fs.WindowsFileCopy.move(WindowsFileCopy.java:387)
>>> > > at
>>> > > sun.nio.fs.WindowsFileSystemProvider.move(WindowsFileSystemProvider.
>>> > java:287)
>>> > > at java.nio.file.Files.move(Files.java:1395)
>>> > > at
>>> > > org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(
>>> > Utils.java:697)
>>> > > at
>>> > > org.apache.kafka.common.record.FileRecords.renameTo(
>>> &

Re: Kafka 2.11-1.1.0 crashes brokers and brings cluster down on Windows

[Ted Yu]

There are two outstanding issues: KAFKA-6059 and KAFKA-6200 which bear some
resemblance.

Can you try to find how the other process uses the file being deleted ?

https://superuser.com/questions/117902/find-out-which-process-is-locking-a-file-or-folder-in-windows
https://www.computerhope.com/issues/ch000714.htm

Cheers

On Sat, May 12, 2018 at 1:42 PM, M. Manna  wrote:

> Hello,
>
> We are still stuck with this issue where 2.11-1.1.0 distro is failing to
> cleanup logs on Windows and brings the entire cluster down one by one.
> Extending the retention hours and sizes don't help because they burden the
> hard drive.
>
> Here is the log
>
> [2018-05-12 21:36:57,673] INFO [Log partition=test-0, dir=C:\kafka1] Rolled
> > new log segment at offset 45 in 105 ms. (kafka.log.Log)
> > [2018-05-12 21:36:57,673] INFO [Log partition=test-0, dir=C:\kafka1]
> > Scheduling log segment [baseOffset 0, size 2290] for deletion.
> > (kafka.log.Log)
> > [2018-05-12 21:36:57,673] ERROR Error while deleting segments for test-0
> > in dir C:\kafka1 (kafka.server.LogDirFailureChannel)
> > java.nio.file.FileSystemException:
> > C:\kafka1\test-0\.log ->
> > C:\kafka1\test-0\.log.deleted: The process cannot
> > access the file because it is being used by another process.
> >
> > at
> > sun.nio.fs.WindowsException.translateToIOException(
> WindowsException.java:86)
> > at
> > sun.nio.fs.WindowsException.rethrowAsIOException(
> WindowsException.java:97)
> > at sun.nio.fs.WindowsFileCopy.move(WindowsFileCopy.java:387)
> > at
> > sun.nio.fs.WindowsFileSystemProvider.move(WindowsFileSystemProvider.
> java:287)
> > at java.nio.file.Files.move(Files.java:1395)
> > at
> > org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(
> Utils.java:697)
> > at
> > org.apache.kafka.common.record.FileRecords.renameTo(
> FileRecords.java:212)
> > at kafka.log.LogSegment.changeFileSuffixes(LogSegment.scala:415)
> > at kafka.log.Log.kafka$log$Log$$asyncDeleteSegment(Log.scala:
> 1601)
> > at kafka.log.Log.kafka$log$Log$$deleteSegment(Log.scala:1588)
> > at
> > kafka.log.Log$$anonfun$deleteSegments$1$$anonfun$
> apply$mcI$sp$1.apply(Log.scala:1170)
> > at
> > kafka.log.Log$$anonfun$deleteSegments$1$$anonfun$
> apply$mcI$sp$1.apply(Log.scala:1170)
> > at
> > scala.collection.mutable.ResizableArray$class.foreach(
> ResizableArray.scala:59)
> > at
> > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
> > at
> > kafka.log.Log$$anonfun$deleteSegments$1.apply$mcI$sp(Log.scala:1170)
> > at kafka.log.Log$$anonfun$deleteSegments$1.apply(Log.scala:1161)
> > at kafka.log.Log$$anonfun$deleteSegments$1.apply(Log.scala:1161)
> > at kafka.log.Log.maybeHandleIOException(Log.scala:1678)
> > at kafka.log.Log.deleteSegments(Log.scala:1161)
> > at kafka.log.Log.deleteOldSegments(Log.scala:1156)
> > at kafka.log.Log.deleteRetentionMsBreachedSegme
> nts(Log.scala:1228)
> > at kafka.log.Log.deleteOldSegments(Log.scala:1222)
> > at
> > kafka.log.LogManager$$anonfun$cleanupLogs$3.apply(LogManager.scala:854)
> > at
> > kafka.log.LogManager$$anonfun$cleanupLogs$3.apply(LogManager.scala:852)
> > at
> > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(
> TraversableLike.scala:733)
> > at scala.collection.immutable.List.foreach(List.scala:392)
> > at
> > scala.collection.TraversableLike$WithFilter.
> foreach(TraversableLike.scala:732)
> > at kafka.log.LogManager.cleanupLogs(LogManager.scala:852)
> > at
> > kafka.log.LogManager$$anonfun$startup$1.apply$mcV$sp(
> LogManager.scala:385)
> > at
> > kafka.utils.KafkaScheduler$$anonfun$1.apply$mcV$sp(
> KafkaScheduler.scala:110)
> > at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:62)
> > at
> > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> > at java.util.concurrent.FutureTask.runAndReset(
> FutureTask.java:308)
> > at
> > java.util.concurrent.ScheduledThreadPoolExecutor$
> ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
> > at
> > java.util.concurrent.ScheduledThreadPoolExecutor$
> ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
> > at
> > java.util.concurrent.ThreadPoolExecutor.runWorker(
> ThreadPoolExecutor.java:1149)
> > at
> > java.util.concurrent.ThreadPoolExecutor$Worker.run(
> ThreadPoolExecutor.java:624)
> > at java.lang.Thread.run(Thread.java:748)
> > Suppressed: java.nio.file.FileSystemException:
> > C:\kafka1\test-0\.log ->
> > C:\kafka1\test-0\.log.deleted: The process cannot
> > access the file because it is being used by another process.
> >
> > at
> > 

Re: Help Needed: Leadership Issue upon Kafka Upgrade (ZooKeeper 3.4.9)

[Ted Yu]

Was there any previous connectivity issue to 1.1.1.143:3888 before the
upgrade ?

I assume you have verified that connectivity between broker and 1.1.1.143
 is Okay.

Which zookeeper release are you running ?

Cheers

On Fri, May 11, 2018 at 3:16 PM, Raghav  wrote:

> Hi
>
> We have a 3 node zk ensemble as well as 3 node Kafka Cluster. They both are
> hosted on the same 3 VMs.
>
> Before Restart
> 1. We were on Kafka 0.10.2.1
>
> After Restart
> 1. We moved to Kafka 1.1
>
> We observe that Kafkas report leadership issues, and for lot of partitions
> Leader is -1. I see some logs in ZK that mainly point towards some
> connectivity issue around restart time.
>
> *We are stuck on this one for a while now, and neither rolling restart of
> ZK is helping. Can you please help or point us how we can debug this.*
>
> *2018-05-11_17:20:49.00305 2018-05-11 17:20:49,002 [myid:1] - INFO
> [WorkerReceiver[myid=1]:FastLeaderElection@600] - Notification: 1 (message
> format version), 1 (n.leader), 0x20112 (n.zxid), 0x1 (n.round), LOOKING
> (n.state), 1 (n.sid), 0x2 (n.peerEpoch) LOOKING (my
> state)2018-05-11_17:20:49.01201
> 2018-05-11 17:20:49,010 [myid:1] - WARN
> [WorkerSender[myid=1]:QuorumCnxManager@400] - Cannot open channel to 2 at
> election address /1.1.1.143:3888
> 
> 2018-05-11_17:20:49.01203 java.net.ConnectException: Connection
> refused
> 2018-05-11_17:20:49.01203   at
> java.net.PlainSocketImpl.socketConnect(Native
> Method)
> 2018-05-11_17:20:49.01203   at java.net
> .AbstractPlainSocketImpl.doConnect(
> AbstractPlainSocketImpl.java:345)
> 2018-05-11_17:20:49.01203   at java.net
> .AbstractPlainSocketImpl.connectToAddress(
> AbstractPlainSocketImpl.java:206)
> 2018-05-11_17:20:49.01204   at java.net
> .AbstractPlainSocketImpl.connect(
> AbstractPlainSocketImpl.java:188)
> 2018-05-11_17:20:49.01204   at
> java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
> 2018-05-11_17:20:49.01204   at
> java.net.Socket.connect(Socket.java:589)
> 2018-05-11_17:20:49.01204   at
> org.apache.zookeeper.server.quorum.QuorumCnxManager.
> connectOne(QuorumCnxManager.java:381)
> 2018-05-11_17:20:49.01204   at
> org.apache.zookeeper.server.quorum.QuorumCnxManager.
> toSend(QuorumCnxManager.java:354)
> 2018-05-11_17:20:49.01205   at
> org.apache.zookeeper.server.quorum.FastLeaderElection$
> Messenger$WorkerSender.process(FastLeaderElection.java:452)
> 2018-05-11_17:20:49.01205   at
> org.apache.zookeeper.server.quorum.FastLeaderElection$
> Messenger$WorkerSender.run(FastLeaderElection.java:433)
> 2018-05-11_17:20:49.01206   at java.lang.Thread.run(Thread.java:745)*
>
> Rag
>

Re: Kafka consumer slowly consume the data

[Ted Yu]

bq. load on the node is increasing tremendously

Can you do some profiling to see where the bottleneck was ?
You can pastebin some stack traces.

Which Kafka release do you use ?

Thanks

On Fri, May 11, 2018 at 6:41 AM, Karthick Kumar 
wrote:

> Hi,
>
> I'm using tomcat nodes for Kafka producer and consumer, Recently I faced
> some issues with it.
> Normally the producer and consumer counts matched in the tomcat nodes.
> After some time the produced data is consumed with delay, I'm not sure
> where to check.
>
> The data which was delayed is dumped over the consumer when it is started.
> At this time, the load on the node is increasing tremendously.
>
>
> So any data that was newly produced at this time is getting lost (it is
> available on the producer but not on the consumer).
>
> --
> With Regards,
> Karthick.K
>

Re: Consulting ReadOnlyKeyValueStore from Processor can lead to deadlock

[Ted Yu]

bq. the docs and CachingKVS behavior could improve

I would agree.

Pointing out the usage of ReadWriteLock and mentioning the
withCachingDisabled()
method in doc would help other developers.

On Thu, May 10, 2018 at 11:21 AM, Steven Schlansker <
sschlans...@opentable.com> wrote:

>
> > On May 10, 2018, at 10:48 AM, Steven Schlansker <
> sschlans...@opentable.com> wrote:
> >
> > But it still remains -- when you go an read that ROKVS documentation, it
> sure
> > doesn't prepare you to this possibility!  And, it's a little frustrating
> that
> > we have to have this 'caching' layer at all -- we already had to add
> >
> >// ensure KTable doesn't delay updates due to buffering in cache
> >kafkaStreamProps.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG,
> 0);
>
> Now that I've said this, it seems that since I last checked we got
> 'Materialized.withCachingDisabled'.
> I'll see if that does what I want...  (I still think the docs and
> CachingKVS behavior could improve, though.)
>
>

Re: Kafka offset problem when using Spark Streaming...

[Ted Yu]

Can you give us more information:

release of Spark and Kafka you're using

anything interesting from broker / client logs - feel free to use pastebin
to pass snippet if needed

Thanks

On Wed, May 9, 2018 at 3:42 AM, Pena Quijada Alexander <
a.penaquij...@reply.it> wrote:

> Hi all,
>
> We're facing some problems with ours Spark Streaming jobs, from yesterday
> we have got the following error into our logs when the jobs fail:
>
> java.lang.AssertionError: assertion failed: Beginning offset 562747 is
> after the ending offset 562743 for topic elk-topic partition 0.
>
> Any help about this issue will be appreciated!
>
> Many thanks in advance for the kind cooperation.
>
> Best regards,
>
> Alex
>
>
>
> 
>
> --
> The information transmitted is intended for the person or entity to which
> it is addressed and may contain confidential and/or privileged material.
> Any review, retransmission, dissemination or other use of, or taking of any
> action in reliance upon, this information by persons or entities other than
> the intended recipient is prohibited. If you received this in error, please
> contact the sender and delete the material from any computer.
>

Re: Kafka-Connect : Invalid value java.sql.SQLException: No suitable driver found for jdbc:mysql://127.0.0.1:3306/connect_test

[Ted Yu]

Have you seen this thread ?

https://stackoverflow.com/questions/49433199/kafka-connect-jdbc-source-connector-not-working-for-microsoft-sql-server

On Tue, May 1, 2018 at 5:52 AM, Jagannath Bilgi 
wrote:

> Hi Team,
> Getting subject error message while trying to  create Kafka-JDBC
> connector. Able to execute initial steps successfully from Kafka Connect
> Tutorial — Confluent Platform
> However it fails on executing below command.
> curl -X POST   -H "Content-Type: application/json"   --data '{ "name":
> "quickstart-jdbc-source", "config": { "connector.class":
> "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": 1,
> "connection.url": "jdbc:mysql://127.0.0.1:3306/
> connect_test?user=confluent=confluent", "mode": "incrementing", "
> incrementing.column.name": "id", "timestamp.column.name": "modified",
> "topic.prefix": "quickstart-jdbc-", "poll.interval.ms": 1000 } }'
> http://192.168.99.100:28082/connectors
> Error message
> {"error_code":400,"message":"Connector configuration is invalid and
> contains the following 2 error(s):\nInvalid value java.sql.SQLException: No
> suitable driver found for jdbc:mysql://127.0.0.1:3306/
> connect_test?user=confluent=confluent for configuration Couldn't
> open connection to jdbc:mysql://127.0.0.1:3306/
> connect_test?user=confluent=confluent\nInvalid value
> java.sql.SQLException: No suitable driver found for jdbc:mysql://
> 127.0.0.1:3306/connect_test?user=confluent=confluent for
> configuration Couldn't open connection to jdbc:mysql://127.0.0.1:3306/
> connect_test?user=confluent=confluent\nYou can also find the
> above list of errors at the endpoint `/{connectorType}/config/
> validate`"}root@default:/#
> Please advise over come the issue.
> Thanks and regards
> Jagannath S Bilgi
>

Re: source code location for KSQL

[Ted Yu]

https://github.com/confluentinc/ksql

FYI

On Fri, Apr 27, 2018 at 4:23 PM, Henry Cai 
wrote:

> I think KSQL is also open sourced, where is the source code location for
> KSQL in github?
>
> Thanks.
>

Re: Re: kafka streams with TimeWindows ,incorrect result

[Ted Yu]

Noticed a typo in streams in subject.
Corrected it in this reply.
 Original message From: 杰 杨  Date: 4/27/18  
1:28 AM  (GMT-08:00) To: 杰 杨 , users  
Subject: Re: Re: kafka steams with TimeWindows ,incorrect result 
and I checked windowStore interface found it has put method not get method.
in one second
the stream had sample key and different value in it.
and I must update key value which store in it.



funk...@live.com

From: funk...@live.com
Date: 2018-04-27 16:08
To: users
Subject: Re: Re: kafka steams with TimeWindows ,incorrect result
Hi:
I don't kown what to do with transform function.
and stream is preapred well
like this at blew
key:
44_14_2018-04-27
value:
CountInfo(start=1,active=0,fresh =0)

there is amount data like that。
how I aggregate it with peer 1 seconds using transform function？



funk...@live.com

From: Guozhang Wang
Date: 2018-04-27 01:50
To: users
Subject: Re: Re: kafka steams with TimeWindows ,incorrect result
Using a control message to flush results to downstream (in your case to the
result db) looks good to me as well.

On Thu, Apr 26, 2018 at 10:49 AM, Guozhang Wang  wrote:

> If you're talking about which store to use in your transform function, it
> should be a windowed store.
>
> You can create such a store with the `Stores` factory, and suppose your
> old code has `windowedBy(TimeWindows.of(6))`, then you can do
>
> `
> windows = TimeWindows.of(6);
>
> Stores.WindowStoreBuilder(
> Stores.persistentWindowStore("Counts"),
> windows.maintainMs(),
>
> windows.segments,
>
> windows.size(),
> true)
>
> )
>
> `
>
>
> Guozhang
>
>
>
> On Thu, Apr 26, 2018 at 4:39 AM, 杰 杨  wrote:
>
>> I return back .
>> Which StateStore could I use for this problem?
>> and another idea .I can send 'flush' message into this topic .
>> when received this message could update results to db.
>> I don't know it's work?
>>
>> 
>> funk...@live.com
>>
>> From: Guozhang Wang
>> Date: 2018-03-12 03:58
>> To: users
>> Subject: Re: Re: kafka steams with TimeWindows ,incorrect result
>> If you want to strictly "only have one output per window", then for now
>> you'd probably implement that logic using a lower-level "transform"
>> function in which you can schedule a punctuate function to send all the
>> results at the end of a window.
>>
>> If you just want to reduce the amount of data to your sink, but your sink
>> can still handle overwritten records of the same key, you can enlarge the
>> cache size via the cache.max.bytes.buffering config.
>>
>> https://kafka.apache.org/documentation/#streamsconfigs
>>
>> On Fri, Mar 9, 2018 at 9:45 PM, 杰 杨  wrote:
>>
>> > thx for your reply!
>> > I see that it is designed to operate on an infinite, unbounded stream of
>> > data.
>> > now I want to process for  unbounded stream but divided by time
>> interval .
>> > so what can I do for doing this ?
>> >
>> > 
>> > funk...@live.com
>> >
>> > From: Guozhang Wang
>> > Date: 2018-03-10 02:50
>> > To: users
>> > Subject: Re: kafka steams with TimeWindows ,incorrect result
>> > Hi Jie,
>> >
>> > This is by design of Kafka Streams, please read this doc for more
>> details
>> > (search for "outputs of the Wordcount application is actually a
>> continuous
>> > stream of updates"):
>> >
>> > https://kafka.apache.org/0110/documentation/streams/quickstart
>> >
>> > Note this semantics applies for both windowed and un-windowed tables.
>> >
>> >
>> > Guozhang
>> >
>> > On Fri, Mar 9, 2018 at 5:36 AM, 杰 杨  wrote:
>> >
>> > > Hi：
>> > > I used TimeWindow for aggregate data in kafka.
>> > >
>> > > this is code snippet ;
>> > >
>> > >   view.flatMap(new MultipleKeyValueMapper(client)
>> > > ).groupByKey(Serialized.with(Serdes.String(),
>> > > Serdes.serdeFrom(new CountInfoSerializer(), new
>> > > CountInfoDeserializer(
>> > > .windowedBy(TimeWindows.of(6)).reduce(new
>> > > Reducer() {
>> > > @Override
>> > > public CountInfo apply(CountInfo value1, CountInfo
>> value2) {
>> > > return new CountInfo(value1.start + value2.start,
>> > > value1.active + value2.active, value1.fresh + value2.fresh);
>> > > }
>> > > }) .toStream(new KeyValueMapper> > > String>() {
>> > > @Override
>> > > public String apply(Windowed key, CountInfo
>> value) {
>> > > return key.key();
>> > > }
>> > > }).print(Printed.toSysOut());
>> > >
>> > > 

Re: log retention bytes and log segment bytes

[Ted Yu]

For log.retention.bytes :
A size-based retention policy for logs. Segments are pruned from the log
unless the remaining segments drop below log.retention.bytes
This config is per partition.

For log.segment.bytes :
The maximum size of a log segment file. When this size is reached a new log
segment will be created.



On Thu, Apr 12, 2018 at 10:10 AM, amit mishra  wrote:

> Hi all ,
>
> I am using kafka 0.10.
>
> log.retention.bytes = 5000
> log.retention.check.interval.ms = 6000
> log.retention.hours = 24
> log.retention.minutes = null
> log.retention.ms = null
> log.roll.hours = 168
> log.roll.jitter.hours = 0
> log.roll.jitter.ms = null
> log.roll.ms = null
> log.segment.bytes = 100
>
> Please let me know what does log.retention.bytes and log.segment.bytes
> denotes ?
>
> Regards,
> amit
>

Re: FetcherLagMetrics is not displaying in JConsole

[Ted Yu]

Can you take a look at KAFKA-6156 and see if your cluster had the same
issue ?

Thanks

On Thu, Apr 12, 2018 at 7:50 AM, ishwar panjari 
wrote:

> HI,
>
> After configuration the kafka and zookeeper. it is not displaying the
> kafka.consumer metrics.
>
>
> I am not able to see the following metrics.
>
> kafka.server:type=FetcherLagMetrics,name=ConsumerLag,clientId=([-.\w]+)
> ,topic=([-.\w]+),partition=([0-9]+)
>
>
> I am using the following version.
> *kafka:*  kafka_2.10-0.10.0.1
> *zookeeper:  *zookeeper-3.4.6
> i have follow this URL to setup:
> https://dzone.com/articles/running-apache-kafka-on-windows-os.
>
> I connected with JConsole and verity but its not displaying the
> "FetcherLagMetrics" nor consumer metrics.
>
>
> any body can guide, why FetcherLagMetrics of consumer is not displaying in
> JConsole?
> Or is there any way to get this metrics?
>
>
>
>
> --
> Thank You
> Ishwar Panjari
> Java Developer
> Cell: +91 8147085870
> Skype: ishwar1987
> Email: panjariish...@gmail.com
>

Re: subscribe kafka user group

[Ted Yu]

See https://kafka.apache.org/contact
 Original message From: "vinly.zhao"  
Date: 4/9/18  5:21 AM  (GMT-08:00) To: users@kafka.apache.org Subject: 
subscribe kafka user group 
Thanks,


*Vincent Zhao*

Re: join 2 topic streams --> to another topic

[Ted Yu]

The blog is an interesting reading material.

There was a minor typo:

bq. an event arrives for either the left of right input stream

'of' above should be 'or'.

Cheers

On Sun, Apr 8, 2018 at 2:04 PM, Matthias J. Sax 
wrote:

> Check out this blog post that explain how the different joins work:
> https://www.confluent.io/blog/crossing-streams-joins-apache-kafka/
>
> It's hard to give a general answer -- it depends on the context of your
> application. Are keys unique? Do you want to get exactly one result or
> should a single stock join with multiple dividends? Do you want Stock
> and Dividend join depending the their timestamps?
>
>
> -Matthias
>
> On 4/8/18 1:34 PM, adrien ruffie wrote:
> > Hello all,
> >
> > I have 2 topics streamed by KStream and one
> KStream
> >
> > I want to merge both object's informations (Stock & Dividend) and send
> to  another topic
> >
> > with for example 
> >
> >
> > The  key of 2 two topic is the same. I need to use, leftJoin,
> merge, KTable, ...
> >
> > what is the best solution ? What do you think ?
> >
> >
> > Moreover, how can I make sure the timestamp are pretty much the same?
> (for merging same key at the same time produced) Example:
> >
> > KStream --> key: "YHOO",  and one timestamp of record
> 2018-04-08 19:56:30
> > KStream --> key: "YHOO",  and one timestamp of record
> 2018-04-08 19:55:27
> >
> >
> > I need a Windowing ?
> >
> >
> > Great thank & best regards,
> >
> >
> > Adrien
> >
>
>

Re: kafka-streams Invalid transition attempted from state READY to state ABORTING_TRANSACTION

[Ted Yu]

https://github.com/apache/kafka/pull/4826

I will fill in JIRA Id once Frederic creates the JIRA.

Cheers

On Wed, Apr 4, 2018 at 4:29 PM, Matthias J. Sax <matth...@confluent.io>
wrote:

> Yes. That looks promising to me. Feel free to open an PR after we have a
> JIRA -- or just create the JIRA right away.
>
> -Matthias
>
> On 4/4/18 2:57 PM, Ted Yu wrote:
> > How about the following change ?
> >
> > diff --git
> > a/streams/src/main/java/org/apache/kafka/streams/processor/internals/
> StreamTask.java
> > b/streams/src/main/java/org/apache/kafka/streams/processor/internals/
> StreamTask.java
> > index 8d6e56a..92bedad 100644
> > ---
> > a/streams/src/main/java/org/apache/kafka/streams/processor/internals/
> StreamTask.java
> > +++
> > b/streams/src/main/java/org/apache/kafka/streams/processor/internals/
> StreamTask.java
> > @@ -482,7 +482,7 @@ public class StreamTask extends AbstractTask
> implements
> > ProcessorNodePunctuator
> >  if (eosEnabled) {
> >  if (!clean) {
> >  try {
> > -if (!isZombie) {
> > +if (!isZombie && transactionInFlight) {
> >  producer.abortTransaction();
> >  }
> >  transactionInFlight = false;
> >
> > On Wed, Apr 4, 2018 at 2:02 PM, Matthias J. Sax <matth...@confluent.io>
> > wrote:
> >
> >> Thanks for reporting this.
> >>
> >> It's indeed a bug in Kafka Streams. It's related to this fix:
> >> https://issues.apache.org/jira/browse/KAFKA-6634 -- the corresponding
> PR
> >> introduces the issue.
> >>
> >> Because, we initialize TX delayed, for your case, we never initialize TX
> >> and thus aborting the TX fails.
> >>
> >> Please open a JIRA for the issue.
> >>
> >> -Matthias
> >>
> >> On 4/4/18 9:32 AM, Ted Yu wrote:
> >>> Looking at isTransitionValid():
> >>>
> >>> case ABORTING_TRANSACTION:
> >>>
> >>> return source == IN_TRANSACTION || source ==
> >>> ABORTABLE_ERROR;
> >>>
> >>> The source state is not supposed to be READY.
> >>>
> >>> I don't see READY in the log you posted.
> >>>
> >>>
> >>> Please consider logging a JIRA where you can attach logs.
> >>>
> >>>
> >>> Cheers
> >>>
> >>>
> >>> On Wed, Apr 4, 2018 at 2:49 AM, Frederic Arno <frederica...@gmail.com>
> >>> wrote:
> >>>
> >>>> Hello,
> >>>>
> >>>> I running tests against kafka-streams 1.1 and get the following stack
> >>>> trace (everything was working alright using kafka-streams 1.0):
> >>>>
> >>>> ERROR org.apache.kafka.streams.processor.internals.
> AssignedStreamsTasks
> >> -
> >>>> stream-thread [feedBuilder-XXX-StreamThread-4] Failed to close stream
> >>>> task, 0_2
> >>>> org.apache.kafka.common.KafkaException: TransactionalId
> >> feedBuilder-0_2:
> >>>> Invalid transition attempted from state READY to state
> >> ABORTING_TRANSACTION
> >>>> at org.apache.kafka.clients.producer.internals.
> TransactionManag
> >>>> er.transitionTo(TransactionManager.java:757)
> >>>> at org.apache.kafka.clients.producer.internals.
> TransactionManag
> >>>> er.transitionTo(TransactionManager.java:751)
> >>>> at org.apache.kafka.clients.producer.internals.
> TransactionManag
> >>>> er.beginAbort(TransactionManager.java:230)
> >>>> at org.apache.kafka.clients.producer.KafkaProducer.
> abortTransac
> >>>> tion(KafkaProducer.java:660)
> >>>> at org.apache.kafka.streams.processor.internals.StreamTask.
> >>>> closeSuspended(StreamTask.java:486)
> >>>> at org.apache.kafka.streams.processor.internals.StreamTask.
> >>>> close(StreamTask.java:546)
> >>>> at org.apache.kafka.streams.processor.internals.
> AssignedTasks.c
> >>>> loseNonRunningTasks(AssignedTasks.java:166)
> >>>> at org.apache.kafka.streams.processor.internals.
> AssignedTasks.
> >>>> suspend(AssignedTasks.java:151)
> >>&g

Re: kafka-streams Invalid transition attempted from state READY to state ABORTING_TRANSACTION

[Ted Yu]

How about the following change ?

diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 8d6e56a..92bedad 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -482,7 +482,7 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator
 if (eosEnabled) {
 if (!clean) {
 try {
-if (!isZombie) {
+if (!isZombie && transactionInFlight) {
 producer.abortTransaction();
 }
 transactionInFlight = false;

On Wed, Apr 4, 2018 at 2:02 PM, Matthias J. Sax <matth...@confluent.io>
wrote:

> Thanks for reporting this.
>
> It's indeed a bug in Kafka Streams. It's related to this fix:
> https://issues.apache.org/jira/browse/KAFKA-6634 -- the corresponding PR
> introduces the issue.
>
> Because, we initialize TX delayed, for your case, we never initialize TX
> and thus aborting the TX fails.
>
> Please open a JIRA for the issue.
>
> -Matthias
>
> On 4/4/18 9:32 AM, Ted Yu wrote:
> > Looking at isTransitionValid():
> >
> > case ABORTING_TRANSACTION:
> >
> > return source == IN_TRANSACTION || source ==
> > ABORTABLE_ERROR;
> >
> > The source state is not supposed to be READY.
> >
> > I don't see READY in the log you posted.
> >
> >
> > Please consider logging a JIRA where you can attach logs.
> >
> >
> > Cheers
> >
> >
> > On Wed, Apr 4, 2018 at 2:49 AM, Frederic Arno <frederica...@gmail.com>
> > wrote:
> >
> >> Hello,
> >>
> >> I running tests against kafka-streams 1.1 and get the following stack
> >> trace (everything was working alright using kafka-streams 1.0):
> >>
> >> ERROR org.apache.kafka.streams.processor.internals.AssignedStreamsTasks
> -
> >> stream-thread [feedBuilder-XXX-StreamThread-4] Failed to close stream
> >> task, 0_2
> >> org.apache.kafka.common.KafkaException: TransactionalId
> feedBuilder-0_2:
> >> Invalid transition attempted from state READY to state
> ABORTING_TRANSACTION
> >> at org.apache.kafka.clients.producer.internals.TransactionManag
> >> er.transitionTo(TransactionManager.java:757)
> >> at org.apache.kafka.clients.producer.internals.TransactionManag
> >> er.transitionTo(TransactionManager.java:751)
> >> at org.apache.kafka.clients.producer.internals.TransactionManag
> >> er.beginAbort(TransactionManager.java:230)
> >> at org.apache.kafka.clients.producer.KafkaProducer.abortTransac
> >> tion(KafkaProducer.java:660)
> >> at org.apache.kafka.streams.processor.internals.StreamTask.
> >> closeSuspended(StreamTask.java:486)
> >> at org.apache.kafka.streams.processor.internals.StreamTask.
> >> close(StreamTask.java:546)
> >> at org.apache.kafka.streams.processor.internals.AssignedTasks.c
> >> loseNonRunningTasks(AssignedTasks.java:166)
> >> at org.apache.kafka.streams.processor.internals.AssignedTasks.
> >> suspend(AssignedTasks.java:151)
> >> at org.apache.kafka.streams.processor.internals.TaskManager.sus
> >> pendTasksAndState(TaskManager.java:242)
> >> at org.apache.kafka.streams.processor.internals.StreamThread$Re
> >> balanceListener.onPartitionsRevoked(StreamThread.java:291)
> >> at org.apache.kafka.clients.consumer.internals.ConsumerCoordina
> >> tor.onJoinPrepare(ConsumerCoordinator.java:414)
> >> at org.apache.kafka.clients.consumer.internals.AbstractCoordina
> >> tor.joinGroupIfNeeded(AbstractCoordinator.java:359)
> >> at org.apache.kafka.clients.consumer.internals.AbstractCoordina
> >> tor.ensureActiveGroup(AbstractCoordinator.java:316)
> >> at org.apache.kafka.clients.consumer.internals.ConsumerCoordina
> >> tor.poll(ConsumerCoordinator.java:290)
> >> at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(
> >> KafkaConsumer.java:1149)
> >> at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaCo
> >> nsumer.java:1115)
> >> at org.apache.kafka.streams.processor.internals.StreamThread.
> >> pollRequests(StreamThread.java:827)
> >> at org.apache.kafka.streams

Re: TimeoutException not being triggered, stuck on OUT_OF_ORDER_SEQUENCE_NUMBER

[Ted Yu]

Which Kafka release are you using ?

I was looking at some potentially related JIRA(s), such as KAFKA-6015.

FYI

On Wed, Apr 4, 2018 at 3:05 AM, Saheb Motiani 
wrote:

> Hi all,
>
> We have been seeing this issue intermittently, and hence it's difficult to
> give a step by step instructions to reproduce it. I have been studying the
> code base of the Sender.java
> (org.apache.kafka.clients.producer.internals.Sender.java), but haven't
> been
> able to find the possible bug.
>
> We are using setup is 3 node Kafka cluster.
>
> Here are some relevant logs:
>
> 2018-03-28 09:50:54,290 ERROR [kafka-producer-network-thread | producer-1]
> o.a.k.c.producer.internals.Sender:301 - [Producer clientId=producer-1] The
> broker returned org.apache.kafka.common.errors.UnknownProducerIdException:
> This exception is raised by the broker if it could not locate the producer
> metadata associated with the producerId in question. This could happen if,
> for instance, the producer's records were deleted because their retention
> time had elapsed. Once the last records of the producerId are removed, the
> producer's metadata is removed from the broker, and future appends by the
> producer will return this exception. for topic-partition pipeline-0 at
> offset -1. This indicates data loss on the broker, and should be
> investigated.
>
> 2018-03-28 09:51:13,394 WARN [kafka-producer-network-thread | producer-1]
> o.a.k.c.producer.internals.Sender:251 - [Producer clientId=producer-1] Got
> error produce response with correlation id 1000 on topic-partition
> pipeline-3, retrying (2147483459 attempts left). Error:
> OUT_OF_ORDER_SEQUENCE_NUMBER
>
> 2018-03-28 10:48:33,365 WARN [kafka-producer-network-thread | producer-1]
> o.a.k.c.producer.internals.Sender:251 - [Producer clientId=producer-1] Got
> error produce response with correlation id 34893 on topic-partition
> pipeline-3, retrying (2147449585 attempts left). Error:
> OUT_OF_ORDER_SEQUENCE_NUMBER
>
> [2018-03-28 09:50:54,421] ERROR [ReplicaManager broker=1001] Error
> processing append operation on partition pipeline-3
> (kafka.server.ReplicaManager)
> org.apache.kafka.common.errors.OutOfOrderSequenceException: Out of order
> sequence number for producerId 5102: 2 (incoming seq. number), 7 (current
> end sequence number)
>
>
> 1. We have some sort of Admin API, which deletes and recreates topics (and
> loads them), and when we delete a topic it creates a new producerId, which
> uses the same producer instance to write messages. (This might be a
> problem, but we don't know for sure)
>
> 2. We don't always get stuck in this INT_MAX retries (because we have
> enabled idempotence), many times it stops after 30 seconds, as expected and
> sets a new producerId. (But sometimes that timeout exception doesn't get
> triggered)
>
> 2018-03-29 10:16:54,826 INFO [kafka-producer-network-thread | producer-1]
> o.a.k.c.p.i.TransactionManager:346 - [Producer clientId=producer-1]
> ProducerId set to -1 with epoch -1
> 2018-03-29 10:16:54,827 INFO [kafka-producer-network-thread | producer-1]
> o.a.k.c.p.i.TransactionManager:346 - [Producer clientId=producer-1]
> ProducerId set to 9002 with epoch 0
>
> ---
> We are looking to eliminate this indeterministic behaviour, by
> handling the OUT_OF_ORDER_SEQUENCE_NUMBER
> in a better way (maybe re-instantiate the producer, but not sure if that
> would solve anything as Kafka has ways to reset producerId after timeout).
>
> Any ideas/comments on why this is happening, regardless of having a default
> timeout of 30 seconds?
>
> Please let me know if you need more information in understanding the
> problem we are facing.
>
> Regards,
> Saheb
> --
> ...
> [image: cake bamtech_logo_rgb signature.jpg]  >
>
> Saheb Motiani
> (Office) 0845 617 1200
> Houldsworth Mill, Houldsworth Street, Reddish, Stockport, SK5 6DA, UK
> www.cakesolutions.net
> [image: twitter-circle-darkgrey.png]
>  [image:
> facebook-circle-darkgrey.png]
>  [image:
> linkedin-circle-darkgrey.png]
> 
> [image: Reactive Applications]
> 
> Company registered in the UK, No. 4184567 If you have received this e-mail
> in error, please accept our apologies, destroy it immediately, and it would
> be greatly appreciated if you notified the sender. It is your
> responsibility to protect your system from viruses and any other harmful
> code or device. We try to eliminate them from e-mails and attachments, but
> we accept no liability for any which remain. We may monitor or access any
> or all e-mails sent to us.
> [image: Powered by Sigstr]
> 
>

Re: kafka-streams Invalid transition attempted from state READY to state ABORTING_TRANSACTION

[Ted Yu]

Looking at isTransitionValid():

case ABORTING_TRANSACTION:

return source == IN_TRANSACTION || source ==
ABORTABLE_ERROR;

The source state is not supposed to be READY.

I don't see READY in the log you posted.


Please consider logging a JIRA where you can attach logs.


Cheers


On Wed, Apr 4, 2018 at 2:49 AM, Frederic Arno 
wrote:

> Hello,
>
> I running tests against kafka-streams 1.1 and get the following stack
> trace (everything was working alright using kafka-streams 1.0):
>
> ERROR org.apache.kafka.streams.processor.internals.AssignedStreamsTasks -
> stream-thread [feedBuilder-XXX-StreamThread-4] Failed to close stream
> task, 0_2
> org.apache.kafka.common.KafkaException: TransactionalId feedBuilder-0_2:
> Invalid transition attempted from state READY to state ABORTING_TRANSACTION
> at org.apache.kafka.clients.producer.internals.TransactionManag
> er.transitionTo(TransactionManager.java:757)
> at org.apache.kafka.clients.producer.internals.TransactionManag
> er.transitionTo(TransactionManager.java:751)
> at org.apache.kafka.clients.producer.internals.TransactionManag
> er.beginAbort(TransactionManager.java:230)
> at org.apache.kafka.clients.producer.KafkaProducer.abortTransac
> tion(KafkaProducer.java:660)
> at org.apache.kafka.streams.processor.internals.StreamTask.
> closeSuspended(StreamTask.java:486)
> at org.apache.kafka.streams.processor.internals.StreamTask.
> close(StreamTask.java:546)
> at org.apache.kafka.streams.processor.internals.AssignedTasks.c
> loseNonRunningTasks(AssignedTasks.java:166)
> at org.apache.kafka.streams.processor.internals.AssignedTasks.
> suspend(AssignedTasks.java:151)
> at org.apache.kafka.streams.processor.internals.TaskManager.sus
> pendTasksAndState(TaskManager.java:242)
> at org.apache.kafka.streams.processor.internals.StreamThread$Re
> balanceListener.onPartitionsRevoked(StreamThread.java:291)
> at org.apache.kafka.clients.consumer.internals.ConsumerCoordina
> tor.onJoinPrepare(ConsumerCoordinator.java:414)
> at org.apache.kafka.clients.consumer.internals.AbstractCoordina
> tor.joinGroupIfNeeded(AbstractCoordinator.java:359)
> at org.apache.kafka.clients.consumer.internals.AbstractCoordina
> tor.ensureActiveGroup(AbstractCoordinator.java:316)
> at org.apache.kafka.clients.consumer.internals.ConsumerCoordina
> tor.poll(ConsumerCoordinator.java:290)
> at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(
> KafkaConsumer.java:1149)
> at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaCo
> nsumer.java:1115)
> at org.apache.kafka.streams.processor.internals.StreamThread.
> pollRequests(StreamThread.java:827)
> at org.apache.kafka.streams.processor.internals.StreamThread.
> runOnce(StreamThread.java:784)
> at org.apache.kafka.streams.processor.internals.StreamThread.
> runLoop(StreamThread.java:750)
> at org.apache.kafka.streams.processor.internals.StreamThread.
> run(StreamThread.java:720)
>
>
> This happens when starting the same stream-processing application on 3
> JVMs all running on the same linux box, JVMs are named JVM-[2-4]. All 3
> instances use separate stream state.dir. No record is ever processed
> because the input kafka topics are empty at this stage.
>
> JVM-2 starts first, joined shortly after by JVM-4 and JVM-3, find the
> state transition logs below. The above stacktrace is from JVM-4
>
> [JVM-2] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from REBALANCING
> to RUNNING
> [JVM-4] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from REBALANCING
> to RUNNING
> [JVM-3] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> JVM-4 crashes here with above stacktrace
> [JVM-2] stream-client [feedBuilder-XXX] State transition from REBALANCING
> to RUNNING
> [JVM-4] stream-client [feedBuilder-XXX] State transition from REBALANCING
> to ERROR
> [JVM-4] stream-client [feedBuilder-XXX] State transition from ERROR to
> PENDING_SHUTDOWN
> [JVM-4] stream-client [feedBuilder-XXX] State transition from
> PENDING_SHUTDOWN to NOT_RUNNING
> [JVM-4] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-3] stream-client [feedBuilder-XXX] State transition from REBALANCING
> to RUNNING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-3] stream-client [feedBuilder-XXX] State transition from RUNNING to
> REBALANCING
> [JVM-2] stream-client [feedBuilder-XXX] State transition from 

Re: kafka-streams TopologyTestDriver problem with EXACTLY_ONCE

[Ted Yu]

You saw the error because TopologyTestDriverTest doesn't explicitly call
MockProducer#initTransactions().

To do that, TopologyTestDriver needs to expose such method to the
TopologyTestDriverTest.

You can log a JIRA if you think adding such capability is needed.

Cheers

On Wed, Apr 4, 2018 at 3:03 AM, Frederic Arno 
wrote:

> Hello,
>
> I can't successfully test my processing application which require
> EXACTLY_ONCE processing guarantee using the new TopologyTestDriver. I
> always get the following exception:
> java.lang.IllegalStateException: MockProducer hasn't been initialized for
> transactions
>
> The tests all work fine as soon as I set PROCESSING_GUARANTEE_CONFIG to
> AT_LEAST_ONCE.
>
> Am I doing something wrong?
>
> Kafka developpers can reproduce this by adding:
> put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
> StreamsConfig.EXACTLY_ONCE);
>
> to line 88 of TopologyTestDriverTest: streams/test-utils/src/test/ja
> va/org/apache/kafka/streams/TopologyTestDriverTest.java
>
> Thanks, Fred
>

Re: SSL does not work

[Ted Yu]

Have you seen this thread ?

http://search-hadoop.com/m/Kafka/uyzND1aC37obDa1W1?subj=Re+advertised+listeners

On Tue, Apr 3, 2018 at 12:28 AM, Eduard  wrote:

> Hey!
>
> I've configured SSL for client connection and it does not work for me:
>
> My configuration:
>
> listeners=PLAINTEXT://0.0.0.0:7032,SSL://0.0.0.0:9093
> security.inter.broker.protocol=PLAINTEXT
> ssl.client.auth=required
>
> ssl.truststore.location=/home/ec2-user/kafka/config/kafka.se
> rver.truststore.jks
> ssl.truststore.password=mypassword
> ssl.keystore.location=/home/ec2-user/kafka/config/kafka.serv
> er.keystore.jks
> ssl.keystore.password=mypassword
> ssl.key.password=mypassword
> advertised.listeners=PLAINTEXT://someaddress:9092,SSL://someaddress:9093
> ssl.enabled.protocols=TLSv1.1
>
> When using openssl client I get following error:
> 139654442704832:error:1408F10B:SSL routines:ssl3_get_record:wrong version
> number:ssl/record/ssl3_record.c:252:
>
> Both kafka consumer and producer work using plaintext on port 9092 but
> don't work using SSL on 9093 port both from local machine and from remote
>

Re: suggestion for 1.1.0 release notes

[Ted Yu]

Nice reminder.

zookeeper community seems to be close to releasing 3.4.12 :

http://search-hadoop.com/m/ZooKeeper/SvKKH1jwLHJq2tr22?subj=Re+VOTE+Apache+ZooKeeper+release+3+4+12+candidate+1

FYI

On Fri, Mar 30, 2018 at 3:38 PM, Gibson, Bill (ArcSight) <
bill.gib...@microfocus.com> wrote:

> The release notes include this resolved issue:
>
> [KAFKA-6390] - Update ZooKeeper to 3.4.11, Gradle and other minor updates
>
> I saw 3.4.10 in the downloaded bits, not 3.4.11. It looks like the update
> to ZooKeeper 3.4.11 was done, but then reverted (commit 825bfe5) because of
> ZOOKEEPER-2960. The reversion is not visible in KAFKA-6390, nor elsewhere
> in release notes.
>
> Since the 1.1.0 release is not affected by the ZK 3.4.11 bug, it might be
> a good idea to say so.  I had to search a bit to find this result.
>
> https://issues.apache.org/jira/browse/KAFKA-6390
> https://github.com/apache/kafka/commit/825bfe5adefe4b5f967068b332cb0f
> eedd7dc4fc
> https://issues.apache.org/jira/browse/ZOOKEEPER-2960
>
>

Re: Apache Kafka / Spark Integration - Exception - The server disconnected before a response was received.

[Ted Yu]

Can you post the stack trace for NetworkException (pastebin) ?

Please also check the broker logs to see if there was some clue around the
time this happened.

Thanks

On Mon, Mar 26, 2018 at 9:30 AM, M Singh 
wrote:

> Hi:
> I am working with spark 2.2.1 and spark kafka 0.10 client integration with
> Kafka brokers using 0.11.
> I get the exception - org.apache.kafka.common.errors.NetworkException:
> The server disconnected before a response was received - when the
> application is trying to write to a topic. This exception kills the spark
> application.
> Based on some similar issues I saw on the web I've added the following
> kafka configuration but it has not helped.
> acks = 0
> request.timeout.ms = 45000
> receive.buffer.bytes = 1024000
> I've posted this question to apache spark users list but have not received
> any response.  If anyone has any suggestion/pointers, please let me know.
> Thanks
>

Re: Running kafka in containers

[Ted Yu]

Hi,Can you give a little more detail on the type and size of the containers ?
Thanks
 Original message From: Thomas Crayford 
 Date: 3/22/18  11:19 AM  (GMT-08:00) To: Users 
 Subject: Re: Running kafka in containers 
We (heroku) have run databases in containers since 2012, and kafka works
just as well as everything else. So: yes

Re: information about kafka

[Ted Yu]

Can you clarify the question ?

Are you looking for different systems for stream processing ?

What's your use case ?

Thanks

On Mon, Mar 19, 2018 at 10:24 AM, abdelhadi FEKHAR  wrote:

> Hi,
>
>in science for collecting data in real time, does it exist only stream
> processing and batch processing ? or there is another systems
>
> Best regards.
>
> --
> Abdelhadi FEKHAR
> Student at the National engineering school for computer science
> option :Information systems & technology
>
>  source=link_campaign=sig-email_content=webmail>
> Garanti
> sans virus. www.avast.com
>  source=link_campaign=sig-email_content=webmail>
> <#DAB4FAD8-2DD7-40BB-A1B8-4E2AA1F9FDF2>
>

Re: Two instances of Kafka consumer reading same partition

[Ted Yu]

Can you attach broker and consumer logs to the JIRA ?

Thanks

On Mon, Mar 19, 2018 at 6:15 AM, Narayan Periwal  wrote:

> Hi,
>
> We are facing an issue with the Kafka consumer,  the new library that got
> introduced in 0.9
>
> We are using Kafka broker 0.10.2.1 and consumer client version is also
> 0.10.2.1
>
> The issue that we have faced is that, after rebalancing, some of the
> partitions gets consumed by 2 instances within a consumer group, leading to
> duplication of the entire partition data. Both the instances continue to
> read the same partition until the next rebalancing, or the restart of those
> clients.
>
> Have incorporated the details in this ticket - KAFKA-6681
> 
>
> Please look at it the earliest
>
> Regards,
> Narayan
>
> --
> _
> The information contained in this communication is intended solely for the
> use of the individual or entity to whom it is addressed and others
> authorized to receive it. It may contain confidential or legally privileged
> information. If you are not the intended recipient you are hereby notified
> that any disclosure, copying, distribution or taking any action in reliance
> on the contents of this information is strictly prohibited and may be
> unlawful. If you have received this communication in error, please notify
> us immediately by responding to this email and then delete it from your
> system. The firm is neither liable for the proper and complete transmission
> of the information contained in this communication nor for any delay in its
> receipt.
>

Re: Kafka Over TLS Error - Failed to send SSL Close message - Broken Pipe

[Ted Yu]

Looking at KAFKA-3702, it is still Open.

FYI

On Thu, Mar 15, 2018 at 5:51 PM, Raghav  wrote:

> I am hitting this issue possible in 10.2.1. Can someone please confirm if
> this issue was fixed in 10.2.1 or not ?
>
> R
>
> On Wed, Jun 7, 2017 at 11:50 AM, IT Consultant <0binarybudd...@gmail.com>
> wrote:
>
> > Hi All ,
> >
> > Thanks a lot for your help .
> >
> > A bug has been logged for said issue and can be found at ,
> >
> > https://issues.apache.org/jira/browse/KAFKA-5401
> >
> >
> > Thanks again .
> >
> > On Sun, Jun 4, 2017 at 6:38 PM, Martin Gainty 
> wrote:
> >
> > >
> > > 
> > > From: IT Consultant <0binarybudd...@gmail.com>
> > > Sent: Friday, June 2, 2017 11:02 AM
> > > To: users@kafka.apache.org
> > > Subject: Kafka Over TLS Error - Failed to send SSL Close message -
> Broken
> > > Pipe
> > >
> > > Hi All,
> > >
> > > I have been seeing below error since three days ,
> > >
> > > Can you please help me understand more about this ,
> > >
> > >
> > > WARN Failed to send SSL Close message
> > > (org.apache.kafka.common.network.SslTransportLayer)
> > > java.io.IOException: Broken pipe
> > >  at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
> > >  at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:
> 47)
> > >  at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
> > >  at sun.nio.ch.IOUtil.write(IOUtil.java:65)
> > >  at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:
> > 471)
> > >  at
> > > org.apache.kafka.common.network.SslTransportLayer.
> > > flush(SslTransportLayer.java:194)
> > >
> > > MG>Here is org.apache.kafka.common.network.SslTransportLayer code:
> > > /**
> > > * Flushes the buffer to the network, non blocking
> > > * @param buf ByteBuffer
> > > * @return boolean true if the buffer has been emptied out, false
> > > otherwise
> > > * @throws IOException
> > > */
> > > private boolean flush(ByteBuffer buf) throws IOException {
> > > int remaining = buf.remaining();
> > > if (remaining > 0) {
> > > int written = socketChannel.write(buf); //no check for
> > > isOpen() *socketChannel.isOpen()*
> > > return written >= remaining;
> > > }
> > > return true;
> > > }
> > >
> > > MG>it appears upstream monitor *container* closed connection but kafka
> > > socketChannel never tested (now-closed) connection with isOpen()
> > > MG>i think you found a bug
> > > MG>can you file bug in kafka-jira ?
> > > https://issues.apache.org/jira/browse/KAFKA/?selectedTab=com.atlassian
> .
> > > jira.jira-projects-plugin:summary-panel
> > > Kafka - ASF JIRA - issues.apache.org > > issues.apache.org/jira/browse/KAFKA/?selectedTab=com.
> > > atlassian.jira.jira-projects-plugin:summary-panel>
> > > issues.apache.org
> > > Atlassian JIRA Project Management Software (v6.3.15#6346-sha1:dbc023d)
> > > About JIRA; Report a problem; Powered by a free Atlassian JIRA open
> > source
> > > license for Apache ...
> > >
> > >
> > >
> > >
> > >  at
> > > org.apache.kafka.common.network.SslTransportLayer.
> > > close(SslTransportLayer.java:148)
> > >  at
> > > org.apache.kafka.common.network.KafkaChannel.close(
> KafkaChannel.java:45)
> > >  at
> > > org.apache.kafka.common.network.Selector.close(Selector.java:442)
> > >  at org.apache.kafka.common.network.Selector.poll(
> > > Selector.java:310)
> > >  at
> > > org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
> > >  at
> > > org.apache.kafka.clients.producer.internals.Sender.run(
> Sender.java:216)
> > >  at
> > > org.apache.kafka.clients.producer.internals.Sender.run(
> Sender.java:128)
> > >  at java.lang.Thread.run(Thread.java:745)
> > >
> > >
> > > Thanks  a lot.
> > >
> >
>
>
>
> --
> Raghav
>

Re: Kafka 0.10.2.2 release

[Ted Yu]

Only found the following from brief search:

http://search-hadoop.com/m/Kafka/uyzND1kSw4Xm3wyj1?subj=0+10+2+2+bug+fix+release

FYI

On Thu, Mar 8, 2018 at 10:13 AM, Devendar Rao 
wrote:

> Hi,
>
> We're hitting an issue with log cleaner and I see this is fixed in 0.10.2.2
> release as per https://issues.apache.org/jira/browse/KAFKA-5413 . But I
> don't see 0.10.2.2 release notes in https://kafka.apache.org/downloads
>
> was 0.10.2.2 ever released?
>
> Thanks,
> Devendar
>

Re: [DISCUSS] KIP-267: Add Processor Unit Test Support to Kafka Streams Test Utils

[Ted Yu]

Looks good.

See if you can add punctuator into the sample code.

On Wed, Mar 7, 2018 at 7:10 PM, John Roesler  wrote:

> Dear Kafka community,
>
> I am proposing KIP-267 to augment the public Streams test utils API.
> The goal is to simplify testing of Kafka Streams applications.
>
> Please find details in the
> wiki:https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 267%3A+Add+Processor+Unit+Test+Support+to+Kafka+Streams+Test+Utils
>
> An initial WIP PR can be found here:https://github.com/
> apache/kafka/pull/4662
>
> I also included the user-list (please hit "reply-all" to include both
> lists in this KIP discussion).
>
> Thanks,
>
> -John
>

Re: [VOTE] 1.1.0 RC1

[Ted Yu]

+1

Checked signature
Ran test suite - apart from flaky testMetricsLeak, other tests passed.

On Tue, Mar 6, 2018 at 2:45 AM, Damian Guy  wrote:

> Hello Kafka users, developers and client-developers,
>
> This is the second candidate for release of Apache Kafka 1.1.0.
>
> This is minor version release of Apache Kakfa. It Includes 29 new KIPs.
> Please see the release plan for more details:
>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=71764913
>
> A few highlights:
>
> * Significant Controller improvements (much faster and session expiration
> edge cases fixed)
> * Data balancing across log directories (JBOD)
> * More efficient replication when the number of partitions is large
> * Dynamic Broker Configs
> * Delegation tokens (KIP-48)
> * Kafka Streams API improvements (KIP-205 / 210 / 220 / 224 / 239)
>
> Release notes for the 1.1.0 release:
> http://home.apache.org/~damianguy/kafka-1.1.0-rc1/RELEASE_NOTES.html
>
> *** Please download, test and vote by Friday, March 9th, 5pm PT
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~damianguy/kafka-1.1.0-rc1/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~damianguy/kafka-1.1.0-rc1/javadoc/
>
> * Tag to be voted upon (off 1.1 branch) is the 1.1.0 tag:
> https://github.com/apache/kafka/tree/1.1.0-rc1
>
>
> * Documentation:
> http://kafka.apache.org/11/documentation.html
>
> * Protocol:
> http://kafka.apache.org/11/protocol.html
>
> * Successful Jenkins builds for the 1.1 branch:
> Unit/integration tests: https://builds.apache.org/job/kafka-1.1-jdk7/68
> System tests: https://jenkins.confluent.io/job/system-test-kafka/job/1.1/
> 30/
>
> /**
>
> Thanks,
> Damian Guy
>

Re: 答复: which Kafka StateStore could I use ?

[Ted Yu]

Jie:
Which DB are you using ?

600 records/second is very low rate.

Probably your DB needs some tuning.

Cheers

On Fri, Mar 2, 2018 at 9:32 AM, Guozhang Wang  wrote:

> Hello Jie,
>
> By default Kafka Streams uses caching on top of its internal state stores
> to de-dup output streams to the final destination (in your case the DB) so
> that for a single key, fewer updates will be generated giving a small
> working set. If your aggregation logic follows such key distribution, you
> can try enlarge the cache size (by default it is only 50MB) and see if it
> helps reduce the downstream traffic to your DB.
>
>
> Guozhang
>
>
> On Thu, Mar 1, 2018 at 6:33 PM, 杰 杨  wrote:
>
> > Yes .but the DB’s Concurrent quantity is  the limitation.
> > Now I can process 600 records/second
> > And I want enhance it
> >
> > 发送自 Windows 10 版邮件应用
> >
> > 发件人: Guozhang Wang
> > 发送时间: 2018年3月2日 2:59
> > 收件人: users@kafka.apache.org
> > 主题: Re: which Kafka StateStore could I use ?
> >
> > Hello Jie,
> >
> > Just to understand your problem better, are you referring "db" for an
> > external storage engine outside Kafka Streams, and you are asking how to
> > only send one record per aggregation key (assuming you are doing some
> > aggregations with Streams' statestore) to that end storage engine?
> >
> >
> > Guozhang
> >
> >
> > On Wed, Feb 28, 2018 at 7:53 PM, 杰 杨  wrote:
> >
> > >
> > > HI：
> > > I use kafka streams for real-time data analysis
> > > and I meet a problem.
> > > now I process a record in kafka and compute it and send to db.
> > > but db concurrency level is not suit for me.
> > > so I want that
> > > 1）when there is not data in kakfa ,the statestore is  no results.
> > > 2) when there is a lot of data records in kafka the statestore save
> > > computed result and I need send its once to db.
> > > which StateStoe can I use for do that above
> > > 
> > > funk...@live.com
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
> >
>
>
> --
> -- Guozhang
>

Re: Kafka Zookeeper Connection

[Ted Yu]

Please take a look at maxSessionTimeout under:
http://zookeeper.apache.org/doc/r3.3.3/zookeeperAdmin.html#sc_advancedConfiguration

On Sat, Feb 24, 2018 at 9:46 AM, Soheil Pourbafrani 
wrote:

> Thanks, Manna, Can you say which property (in Kafka or Zookeeper) should I
> increase?
>
> On Sat, Feb 24, 2018 at 7:49 PM, M. Manna  wrote:
>
> > Have you tried increasing the timeouts for zookeeper nodes and Kafka
> > brokers to see if they make a difference?
> >
> >
> >
> > On Sat, 24 Feb 2018 at 14:55, Soheil Pourbafrani 
> > wrote:
> >
> > > Hi and Thanks,
> > > Excuse me, The Kafka version is 0.11 and Zookeeper version is 3.4.10
> > >
> > > I've check Zookeeper logs and sessions are expiring and renewing
> > > continuously there. I use the same Zookeeper cluster for Hadoop HA and
> it
> > > works well.
> > >
> > > I Answer the same question with more details in the following link:
> > >
> > > Here
> > > <
> > > https://stackoverflow.com/questions/48885469/kafka-
> > zookeeper-connection-drop-continuously
> > > >
> > >
> >
>

Re: Kafka Zookeeper Connection

[Ted Yu]

Have you seen this thread (GC might be the cause for session time out) ?

http://search-hadoop.com/m/Kafka/uyzND15d4h21yZnIZ?subj=Re+zookeeper+session+time+out

Please check zookeeper server logs when you have a chance.

Please share the versions of Kafka and zookeeper.

On Sat, Feb 24, 2018 at 12:03 AM, Soheil Pourbafrani 
wrote:

> In Kafka 0.11 server.log, it continuously logs
>
> [2018-02-24 10:38:11,178] WARN Client session timed out, have not heard
> from server in 4001ms for sessionid 0x361c65434f1000c
> (org.apache.zookeeper.ClientCnxn)
> [2018-02-24 10:38:11,178] INFO Client session timed out, have not heard
> from server in 4001ms for sessionid 0x361c65434f1000c, closing socket
> connection and attempting$
> [2018-02-24 10:38:11,278] INFO zookeeper state changed (Disconnected)
> (org.I0Itec.zkclient.ZkClient)
> [2018-02-24 10:38:11,979] INFO Opening socket connection to server
> zookeeper1/192.168.1.204:2181. Will not attempt to authenticate using SASL
> (unknown error) (org.apac$
> [2018-02-24 10:38:12,981] INFO Socket connection established to zookeeper1/
> 192.168.1.204:2181, initiating session (org.apache.zookeeper.ClientCnxn)
> [2018-02-24 10:38:12,990] INFO Session establishment complete on server
> zookeeper1/192.168.1.204:2181, sessionid = 0x361c65434f1000c, negotiated
> timeout = 6000 (org.ap$
> [2018-02-24 10:38:12,990] INFO zookeeper state changed (SyncConnected)
> (org.I0Itec.zkclient.ZkClient)
>
> But Kafka work and for example, I can get topic list using kafka-topic
> script
> My Kafka cluster contains 3 nodes and Zookeeper cluster contains 3 nodes,
> too (but in different machines).
>
> The following is Kafka conf:
>
> broker.id=100
> listeners=PLAINTEXT://kafka1:9092
> num.partitions=24
> delete.topic.enable=true
> default.replication.factor=3
> log.dirs=/data/kafka/data
> zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181
> log.retention.hours=168
>
> Can anyone help?
>

Re: Testing with MockConsumer

[Ted Yu]

For #3, a better example would be in ConsumerCoordinator (around line 632).

commitOffsetsAsync(allConsumedOffsets, new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata>
offsets, Exception exception) {

FYI

On Mon, Feb 19, 2018 at 10:56 AM, Gabriel Giussi <gabrielgiu...@gmail.com>
wrote:

> Hi Ted,
> my mistake was believe that commited offsets are used on the next poll, but
> is not the case
> <https://github.com/apache/kafka/blob/73be1e1168f91ee2a9d68e1d1c75c1
> 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> consumer/KafkaConsumer.java#L1202>
> .
>
> > The offsets committed using this API will be used on the first fetch
> after
> > every rebalance and also on startup
> >
>
> So, what to do after a failed commit depends on the nature of the exception
> I guess.
>
>- WakeupException: retry
>- Others: close consumer
>
> Thanks for your help to solve #2. I'm wondering about 1# and 3# yet.
>
> 2018-02-19 11:46 GMT-03:00 Ted Yu <yuzhih...@gmail.com>:
>
> > For #2, I think the assumption is that the records are processed by the
> > loop:
> >
> > https://github.com/apache/kafka/blob/73be1e1168f91ee2a9d68e1d1c75c1
> > 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> > consumer/MockConsumer.java#L164
> >
> >
> >
> > On Mon, Feb 19, 2018 at 4:39 AM, Gabriel Giussi <gabrielgiu...@gmail.com
> >
> > wrote:
> >
> > > Hi,
> > >
> > > I'm trying to use MockConsumer to test my application code but I've
> > faced a
> > > couple of limitations and I want to know if there are workarounds or
> > > something that I'm overlooking.
> > > Note: I'm using kafka-clients v 0.11.0.2
> > >
> > >
> > >1. Why the addRecord
> > ><https://github.com/apache/kafka/blob/
> 73be1e1168f91ee2a9d68e1d1c75c1
> > > 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> > > consumer/MockConsumer.java#L179>
> > >requires that the consumer has assigned partitions? Given that this
> is
> > > just
> > >simulating records being produced or existing records.
> > >2. Why the poll
> > ><https://github.com/apache/kafka/blob/
> 73be1e1168f91ee2a9d68e1d1c75c1
> > > 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> > > consumer/MockConsumer.java#L132>
> > >clear the map of records? It should not be cleared after commit?
> > >3. Why the commitAsync
> > ><https://github.com/apache/kafka/blob/
> 73be1e1168f91ee2a9d68e1d1c75c1
> > > 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> > > consumer/MockConsumer.java#L198>
> > >doesn't check for an exception and always succeed?
> > >
> > > Due to items (2) and (3) I'm not be able to test scenarios where the
> > > commits fails and the consumer should poll again the same elements.
> > >
> > > If someone knows about other scenarios that can't be tested with
> > > MockConsumer, please let me know.
> > >
> > > Thanks.
> > >
> >
>

Re: Testing with MockConsumer

[Ted Yu]

For #2, I think the assumption is that the records are processed by the
loop:

https://github.com/apache/kafka/blob/73be1e1168f91ee2a9d68e1d1c75c14018cf7d3a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java#L164



On Mon, Feb 19, 2018 at 4:39 AM, Gabriel Giussi 
wrote:

> Hi,
>
> I'm trying to use MockConsumer to test my application code but I've faced a
> couple of limitations and I want to know if there are workarounds or
> something that I'm overlooking.
> Note: I'm using kafka-clients v 0.11.0.2
>
>
>1. Why the addRecord
> 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> consumer/MockConsumer.java#L179>
>requires that the consumer has assigned partitions? Given that this is
> just
>simulating records being produced or existing records.
>2. Why the poll
> 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> consumer/MockConsumer.java#L132>
>clear the map of records? It should not be cleared after commit?
>3. Why the commitAsync
> 4018cf7d3a/clients/src/main/java/org/apache/kafka/clients/
> consumer/MockConsumer.java#L198>
>doesn't check for an exception and always succeed?
>
> Due to items (2) and (3) I'm not be able to test scenarios where the
> commits fails and the consumer should poll again the same elements.
>
> If someone knows about other scenarios that can't be tested with
> MockConsumer, please let me know.
>
> Thanks.
>

Re: KafkaUtils.createStream(..) is removed for API

[Ted Yu]

createStream() is still
in 
external/kafka-0-8/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala
But it is not
in 
external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala

FYI

On Sun, Feb 18, 2018 at 5:17 PM, naresh Goud 
wrote:

> Hello Team,
>
> I see "KafkaUtils.createStream() " method not available in spark 2.2.1.
>
> Can someone please confirm if these methods are removed?
>
> below is my pom.xml entries.
>
>
> 
>   2.11.8
>   2.11
> 
>
>
>   
>   org.apache.spark
>   spark-streaming_${scala.tools.version}
>   2.2.1
>   provided
>   
> 
>   org.apache.spark
>   spark-streaming-kafka-0-10_2.11
>   2.2.1
>   provided
> 
> 
>   org.apache.spark
>   spark-core_2.11
>   2.2.1
>   provided
> 
>   
>
>
>
>
>
> Thank you,
> Naresh
>

Re: Zookeeper Error

[Ted Yu]

What are the entries in /etc/hosts w.r.t. localhost ?

I wonder if the exception had something to do with ipv6.

On Sat, Feb 17, 2018 at 5:49 PM, Maria Pilar  wrote:

> When i try to create a topic in that multicluster,
>
> kafka-topics.bat --create --topic my-kafka-topic --zookeeper locahost:2181
> --replication-factor 2 --partitions 3
>
>
> i have received the same error
>
> Exception in thread "main" org.I0Itec.zkclient.exception.ZkException:
> Unable to connect to locahost:2181
> at org.I0Itec.zkclient.ZkConnection.connect(ZkConnection.java:72)
> at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1228)
> at org.I0Itec.zkclient.ZkClient.(ZkClient.java:157)
> at org.I0Itec.zkclient.ZkClient.(ZkClient.java:131)
> at
> kafka.utils.ZkUtils$.createZkClientAndConnection(ZkUtils.scala:115)
> at kafka.utils.ZkUtils$.apply(ZkUtils.scala:97)
> at kafka.admin.TopicCommand$.main(TopicCommand.scala:56)
> at kafka.admin.TopicCommand.main(TopicCommand.scala)
> Caused by: java.net.UnknownHostException: locahost
> at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
> at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:928)
> at
> java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1323)
> at java.net.InetAddress.getAllByName0(InetAddress.java:1276)
> at java.net.InetAddress.getAllByName(InetAddress.java:1192)
> at java.net.InetAddress.getAllByName(InetAddress.java:1126)
> at
> org.apache.zookeeper.client.StaticHostProvider.(
> StaticHostProvider.java:61)
> at org.apache.zookeeper.ZooKeeper.(ZooKeeper.java:445)
> at org.apache.zookeeper.ZooKeeper.(ZooKeeper.java:380)
> at org.I0Itec.zkclient.ZkConnection.connect(ZkConnection.java:70)
> ... 7 more
>
> Thanks
>
> On 18 February 2018 at 02:45, Maria Pilar  wrote:
>
> > Hi
> >
> > I´m trying to configure a multinode cluster in kafka. I have configured
> > each server.properties according with the new properties for each server.
> >
> > When i start each server, the zookeeper console shows that error.
> >
> >  INFO Got user-level KeeperException when processing
> > sessionid:0x161a690f731 type:create cxid:0xd8 zxid:0x11f txntype:-1
> > reqpath:n/a Error Path:/brokers/ids Error:KeeperErrorCode = NodeExists
> for
> > /brokers/ids (org.apache.zookeeper.server.PrepRequestProcessor)
> >
> > I have configured each broker.id
> >
> >
> > Anybody knows what is the error?
> >
> > Thanks
> >
>

Re: Kafka connect mysql

[Ted Yu]

In your first email, there are two pairs of brackets following
connector-plugins <http://localhost:8083/connector-plugins>
They were output of the curl command, right ?

On page 146, commands for Mac were given.

What commands did you use to install mysql, etc on Debian ?

Thanks

On Sat, Feb 17, 2018 at 2:51 PM, adrien ruffie <adriennolar...@hotmail.fr>
wrote:

> yes the fact that my jdbcSourceConnector wasn't installd 
>
>
> By checking with the REST api connect list and status, no connector was
> deployed
>
> ____________
> De : Ted Yu <yuzhih...@gmail.com>
> Envoyé : samedi 17 février 2018 23:46:18
> À : users@kafka.apache.org
> Objet : Re: Kafka connect mysql
>
> From your first email: {"class":"io.confluent.connect.jdbc.
> JdbcSourceConnector","type":"source","version":"4.1.0-SNAPSHOT"}
>
> It seems the JdbcSourceConnector was installed.
>
> Did you encounter any other error apart from the globbing complaint ?
>
> Cheers
>
> On Sat, Feb 17, 2018 at 1:59 PM, adrien ruffie <adriennolar...@hotmail.fr>
> wrote:
>
> > yes like suggested :-) but nothing,
> >
> > Debian 9 for the OS
> >
> >
> > thx Ted
> >
> > 
> > De : Ted Yu <yuzhih...@gmail.com>
> > Envoyé : samedi 17 février 2018 22:10:19
> > À : users@kafka.apache.org
> > Objet : Re: Kafka connect mysql
> >
> > Have you tried adding -g/--globoff ?
> >
> > What is the OS you use ?
> >
> > Cheers
> >
> > On Sat, Feb 17, 2018 at 11:04 AM, adrien ruffie <
> adriennolar...@hotmail.fr
> > >
> > wrote:
> >
> > > Hello all,
> > >
> > >
> > > In Kafka the definitive guide, on page 146 I found the following
> command:
> > >
> > >
> > > curl http://localhost:8083/connector-plugins
> > [{"class":"org.apache.kafka.
> > > connect.file.FileStreamSourceConnector"},{"
> class":"io.confluent.connect.
> > > elasticsearch.ElasticsearchSinkConnector"},{"class":"org.apache.kafka.
> > > connect.file.FileStreamSinkConnector"},{"class":"io.confluent.connect.
> > > jdbc.JdbcSourceConnector"}]
> > >
> > > But when I try it, and its return me the following line: curl: (3)
> > > [globbing] bad range specification in column 2
> > >
> > >
> > > ~/Java/kafka_2.11-1.0.0$ curl http://localhost:8083/connector-plugins
> > > [{"class":"org.apache.kafka.connect.file.
> FileStreamSourceConnector"},{"
> > > class":"io.confluent.connect.elasticsearch.
> > ElasticsearchSinkConnector"},{
> > > "class":"org.apache.kafka.connect.file.FileStreamSinkConnector"},{"
> > > class":"io.confluent.connect.jdbc.JdbcSourceConnector"}]
> > > [{"class":"io.confluent.connect.elasticsearch.
> > > ElasticsearchSinkConnector","type":"sink","version":"4.1.0-
> > > SNAPSHOT"},{"class":"io.confluent.connect.jdbc.
> > JdbcSinkConnector","type":"
> > > sink","version":"4.1.0-SNAPSHOT"},{"class":"io.confluent.connect.jdbc.
> > > JdbcSourceConnector","type":"source","version":"4.1.0-
> > > SNAPSHOT"},{"class":"org.apache.kafka.connect.file.
> > > FileStreamSinkConnector","type":"sink","version":"1.0.0"
> > > },{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","
> > > type":"source","version":"1.0.0"}]curl: (3) [globbing] bad range
> > > specification in column 2
> > >
> > >
> > > And it seems that my jdbc connector is not deployed ... do you have an
> > > idea ?
> > >
> > > Best regars,
> > > Adrien
> > >
> > >
> >
>

Re: Kafka connect mysql

[Ted Yu]

>From your first email: {"class":"io.confluent.connect.jdbc.
JdbcSourceConnector","type":"source","version":"4.1.0-SNAPSHOT"}

It seems the JdbcSourceConnector was installed.

Did you encounter any other error apart from the globbing complaint ?

Cheers

On Sat, Feb 17, 2018 at 1:59 PM, adrien ruffie <adriennolar...@hotmail.fr>
wrote:

> yes like suggested :-) but nothing,
>
> Debian 9 for the OS
>
>
> thx Ted
>
> 
> De : Ted Yu <yuzhih...@gmail.com>
> Envoyé : samedi 17 février 2018 22:10:19
> À : users@kafka.apache.org
> Objet : Re: Kafka connect mysql
>
> Have you tried adding -g/--globoff ?
>
> What is the OS you use ?
>
> Cheers
>
> On Sat, Feb 17, 2018 at 11:04 AM, adrien ruffie <adriennolar...@hotmail.fr
> >
> wrote:
>
> > Hello all,
> >
> >
> > In Kafka the definitive guide, on page 146 I found the following command:
> >
> >
> > curl http://localhost:8083/connector-plugins
> [{"class":"org.apache.kafka.
> > connect.file.FileStreamSourceConnector"},{"class":"io.confluent.connect.
> > elasticsearch.ElasticsearchSinkConnector"},{"class":"org.apache.kafka.
> > connect.file.FileStreamSinkConnector"},{"class":"io.confluent.connect.
> > jdbc.JdbcSourceConnector"}]
> >
> > But when I try it, and its return me the following line: curl: (3)
> > [globbing] bad range specification in column 2
> >
> >
> > ~/Java/kafka_2.11-1.0.0$ curl http://localhost:8083/connector-plugins
> > [{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector"},{"
> > class":"io.confluent.connect.elasticsearch.
> ElasticsearchSinkConnector"},{
> > "class":"org.apache.kafka.connect.file.FileStreamSinkConnector"},{"
> > class":"io.confluent.connect.jdbc.JdbcSourceConnector"}]
> > [{"class":"io.confluent.connect.elasticsearch.
> > ElasticsearchSinkConnector","type":"sink","version":"4.1.0-
> > SNAPSHOT"},{"class":"io.confluent.connect.jdbc.
> JdbcSinkConnector","type":"
> > sink","version":"4.1.0-SNAPSHOT"},{"class":"io.confluent.connect.jdbc.
> > JdbcSourceConnector","type":"source","version":"4.1.0-
> > SNAPSHOT"},{"class":"org.apache.kafka.connect.file.
> > FileStreamSinkConnector","type":"sink","version":"1.0.0"
> > },{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","
> > type":"source","version":"1.0.0"}]curl: (3) [globbing] bad range
> > specification in column 2
> >
> >
> > And it seems that my jdbc connector is not deployed ... do you have an
> > idea ?
> >
> > Best regars,
> > Adrien
> >
> >
>

Re: Kafka connect mysql

[Ted Yu]

Have you tried adding -g/--globoff ?

What is the OS you use ?

Cheers

On Sat, Feb 17, 2018 at 11:04 AM, adrien ruffie 
wrote:

> Hello all,
>
>
> In Kafka the definitive guide, on page 146 I found the following command:
>
>
> curl http://localhost:8083/connector-plugins [{"class":"org.apache.kafka.
> connect.file.FileStreamSourceConnector"},{"class":"io.confluent.connect.
> elasticsearch.ElasticsearchSinkConnector"},{"class":"org.apache.kafka.
> connect.file.FileStreamSinkConnector"},{"class":"io.confluent.connect.
> jdbc.JdbcSourceConnector"}]
>
> But when I try it, and its return me the following line: curl: (3)
> [globbing] bad range specification in column 2
>
>
> ~/Java/kafka_2.11-1.0.0$ curl http://localhost:8083/connector-plugins
> [{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector"},{"
> class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"},{
> "class":"org.apache.kafka.connect.file.FileStreamSinkConnector"},{"
> class":"io.confluent.connect.jdbc.JdbcSourceConnector"}]
> [{"class":"io.confluent.connect.elasticsearch.
> ElasticsearchSinkConnector","type":"sink","version":"4.1.0-
> SNAPSHOT"},{"class":"io.confluent.connect.jdbc.JdbcSinkConnector","type":"
> sink","version":"4.1.0-SNAPSHOT"},{"class":"io.confluent.connect.jdbc.
> JdbcSourceConnector","type":"source","version":"4.1.0-
> SNAPSHOT"},{"class":"org.apache.kafka.connect.file.
> FileStreamSinkConnector","type":"sink","version":"1.0.0"
> },{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","
> type":"source","version":"1.0.0"}]curl: (3) [globbing] bad range
> specification in column 2
>
>
> And it seems that my jdbc connector is not deployed ... do you have an
> idea ?
>
> Best regars,
> Adrien
>
>

Re: Kafka cluster instablility

[Ted Yu]

For #2 and #3, you would get better stability if zookeeper and Kafka get
dedicated machines.

Have you profiled the performance of the nodes where multiple processes ran
(zookeeper / Kafka / Druid) ? How was disk and network IO like ?

Cheers

On Wed, Feb 14, 2018 at 9:38 AM, Avinash Herle 
wrote:

> Hi,
>
> I'm using Kafka version 0.11.0.2. In my cluster, I've 4 nodes running Kafka
> of which 3 nodes also running Zookeeper. I've a few producer processes that
> publish to Kafka and multiple consumer processes, a streaming engine
> (Spark) that ingests from Kafka and also publishes data to Kafka, and a
> distributed data store (Druid) which reads all messages from Kafka and
> stores in the DB. Druid also uses the same Zookeeper cluster being used by
> Kafka for cluster state management.
>
> *Kafka Configs:*
> 1) No replication being used
> 2) Number of network threads 30
> 3) Number of IO threads 8
> 4) Machines have 64GB RAM and 16 cores
> 5) 3 topics with 64 partitions per topic
>
> *Questions:*
>
> 1) *Partitions going offline*
> I frequently see partitions going offline because of which the scheduling
> delay of the Spark application increases and input rate gets jittery. I
> tried enabling replication too to see if it helped with the problem. It
> didn't quite make a difference. What could be the cause of this issue? Lack
> of resources or cluster misconfigurations? Can the cause be large number of
> receiver processes?
>
> *2) Colocation of Zookeeper and Kafka:*
> As I mentioned above, I'm running 3 nodes with both Zookeeper and Kafka
> colocated. Both the components are containerized, so they are running
> inside docker containers. I found a few blogs that suggested not colocating
> them for performance reasons. Is it necessary to run them on dedicated
> machines?
>
> *3) Using same Zookeeper cluster across different components*
> In my cluster, I use the same Zookeeper cluster for state management of the
> Kafka cluster and the Druid cluster. Could this cause instability of the
> overall system?
>
> Hope I've covered all the necessary information needed. Please let me know
> if more information about my cluster is needed.
>
> Thanks in advance,
> Avinash
> --
>
> Excuse brevity and typos. Sent from mobile device.
>

Re: error when attempting a unit test of spring kafka producer

[Ted Yu]

LoginType was in 0.10.x release.

This seems to indicate Kafka version mismatch.

Can you check the dependencies of your test ?

Thanks

On Tue, Feb 13, 2018 at 8:03 PM, Ian Ewing  wrote:

> I have been trying to figure out how to unit test a kafka producer. Should
> take in a simple integer and perform some addition. Followed what I could
> find on spring kafka unit testing but keep running into this error:
>
> 19:53:12.788 [main] ERROR kafka.server.KafkaServer - [Kafka Server 0],
> Fatal error during KafkaServer startup. Prepare to shutdown
> java.lang.NoClassDefFoundError: org/apache/kafka/common/network/LoginType
> at kafka.network.Processor.(SocketServer.scala:406)
> at kafka.network.SocketServer.newProcessor(SocketServer.scala:141)
> at
> kafka.network.SocketServer$$anonfun$startup$1$$anonfun$
> apply$1.apply$mcVI$sp(SocketServer.scala:94)
> at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)
> at
> kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:93)
> at
> kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:89)
> at scala.collection.Iterator$class.foreach(Iterator.scala:893)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
> at scala.collection.MapLike$DefaultValuesIterable.foreach(
> MapLike.scala:206)
> at kafka.network.SocketServer.startup(SocketServer.scala:89)
> at kafka.server.KafkaServer.startup(KafkaServer.scala:219)
> at kafka.utils.TestUtils$.createServer(TestUtils.scala:120)
> at kafka.utils.TestUtils.createServer(TestUtils.scala)
> at
> org.springframework.kafka.test.rule.KafkaEmbedded.
> before(KafkaEmbedded.java:154)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:46)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(
> SpringJUnit4ClassRunner.java:191)
> at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
> at
> com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(
> JUnit4IdeaTestRunner.java:68)
> at
> com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.
> startRunnerWithArgs(IdeaTestRunner.java:51)
> at
> com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(
> JUnitStarter.java:242)
> at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
> Caused by: java.lang.ClassNotFoundException:
> org.apache.kafka.common.network.LoginType
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> ... 23 common frames omitted
>
>
> Has anyone come across this situation? Any ideas on the direction a
> solution would take? I can provide more information, code, etc. Whatever
> extra is needed. Don't want to bog down the email with too much.
>
> Thanks
> Ian
>

Typo in "Processing with local state"

[Ted Yu]

Hi,
In "Kafka the definitive guide", page 257:

to calculate the minimum and average price ...

It seems average should be maximum.

Cheers

Re: org.apache.kafka.clients.consumer.OffsetOutOfRangeException

[Ted Yu]

Have you looked at SPARK-19888 ?

Please give the full stack trace of the exception you saw.

Cheers

On Mon, Feb 12, 2018 at 12:38 PM, Mina Aslani  wrote:

> Hi Matthias,
> Are you referring to https://issues.apache.org/jira/browse/SPARK-19976?
> Doesn't look like that the jira was not fixed. (e.g. Resolution: "Not a
> Problem").
> So, is there any suggested workaround?
>
> Regards,
> Mina
>
> On Mon, Feb 12, 2018 at 3:03 PM, Matthias J. Sax 
> wrote:
>
> > AFAIK, Spark does not pass this config to the consumer on purpose...
> > It's not a Kafka issues -- IIRC, there is Spark JIRA ticket for this.
> >
> > -Matthias
> >
> > On 2/12/18 11:04 AM, Mina Aslani wrote:
> > > Hi,
> > >
> > > I am getting below error
> > > Caused by: org.apache.kafka.clients.consumer.
> OffsetOutOfRangeException:
> > > Offsets out of range with no configured reset policy for partitions:
> > > {topic1-0=304337}
> > > as soon as I submit a spark app to my cluster.
> > >
> > > I am using below dependency
> > > name: 'spark-streaming-kafka-0-10_2.11', version: '2.2.0' And setting
> > the
> > > consumer's reset config(e.g. AUTO_OFFSET_RESET_CONFIG) to "earliest".
> > > As per https://kafka.apache.org/0110/documentation.html the exception
> > > should be thrown only when the consumer's reset config has not been set
> > > (e.g. default=none).
> > > Wondering what is the cause and how to fix.
> > >
> > > Best regards,
> > > Mina
> > >
> >
> >
>

Re: [VOTE] 1.0.1 RC1

[Ted Yu]

+1

Ran test suite which passed.

BTW it seems the staging repo hasn't been updated yet:

https://repository.apache.org/content/groups/staging/org/apache/kafka/kafka-clients/

On Mon, Feb 12, 2018 at 10:16 AM, Ewen Cheslack-Postava 
wrote:

> And of course I'm +1 since I've already done normal release validation
> before posting this.
>
> -Ewen
>
> On Mon, Feb 12, 2018 at 10:15 AM, Ewen Cheslack-Postava  >
> wrote:
>
> > Hello Kafka users, developers and client-developers,
> >
> > This is the second candidate for release of Apache Kafka 1.0.1.
> >
> > This is a bugfix release for the 1.0 branch that was first released with
> > 1.0.0 about 3 months ago. We've fixed 49 significant issues since that
> > release. Most of these are non-critical, but in aggregate these fixes
> will
> > have significant impact. A few of the more significant fixes include:
> >
> > * KAFKA-6277: Make loadClass thread-safe for class loaders of Connect
> > plugins
> > * KAFKA-6185: Selector memory leak with high likelihood of OOM in case of
> > down conversion
> > * KAFKA-6269: KTable state restore fails after rebalance
> > * KAFKA-6190: GlobalKTable never finishes restoring when consuming
> > transactional messages
> > * KAFKA-6529: Stop file descriptor leak when client disconnects with
> > staged receives
> >
> > Release notes for the 1.0.1 release:
> > http://home.apache.org/~ewencp/kafka-1.0.1-rc1/RELEASE_NOTES.html
> >
> > *** Please download, test and vote by Thursday, Feb 15, 5pm PT ***
> >
> > Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > * Release artifacts to be voted upon (source and binary):
> > http://home.apache.org/~ewencp/kafka-1.0.1-rc1/
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Javadoc:
> > http://home.apache.org/~ewencp/kafka-1.0.1-rc1/javadoc/
> >
> > * Tag to be voted upon (off 1.0 branch) is the 1.0.1 tag:
> > https://github.com/apache/kafka/tree/1.0.1-rc1
> >
> > * Documentation:
> > http://kafka.apache.org/10/documentation.html
> >
> > * Protocol:
> > http://kafka.apache.org/10/protocol.html
> >
> >
> > Thanks,
> > Ewen Cheslack-Postava
> >
>

Re: DumpLogSegment

[Ted Yu]

I think this was due to the type of file you fed to the tool.
To use --index-sanity-check , you need to supply file with the following
suffix:

  val IndexFileSuffix = ".index"


On Sat, Feb 10, 2018 at 2:09 PM, adrien ruffie <adriennolar...@hotmail.fr>
wrote:

> No really ... just the same output like this:
>
>
> kafka_2.11-1.0.0$ ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments
> --files logs/kafka-1/customer-topic-0/.log
> --index-sanity-check
> Dumping logs/kafka-1/customer-topic-0/.log
> Starting offset: 0
> baseOffset: 0 lastOffset: 0 baseSequence: -1 lastSequence: -1 producerId:
> -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false
> position: 0 CreateTime: 1518299032960 isvalid: true size: 91 magic: 2
> compresscodec: NONE crc: 3301617258
>
>
> moreover you can see my release 2.11-1.0.0
>
>
> thank for you response Ted !
>
> 
> De : Ted Yu <yuzhih...@gmail.com>
> Envoyé : samedi 10 février 2018 23:00:02
> À : users@kafka.apache.org
> Objet : Re: DumpLogSegment
>
> For --index-sanity-check, according to dumpIndex():
>
> if (indexSanityOnly) {
>
>   index.sanityCheck
>
>   println(s"$file passed sanity check.")
>
>
> Do you see the print above ?
>
>
> Which release of Kafka are you using ?
>
>
> Cheers
>
>
> On Sat, Feb 10, 2018 at 1:54 PM, adrien ruffie <adriennolar...@hotmail.fr>
> wrote:
>
> > Hi all,
> >
> > In Kafka the definitive guide in page 200-201, two parameters of
> > kafka.tools.DumpLogSegments appear not really work ...
> >
> > the --index-sanity-check argument
> >
> > the --print-data-log
> >
> > exemple:
> >
> > ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> > logs/kafka-0/customer-topic-0/.log
> > --index-sanity-check
> >
> > ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> > logs/kafka-0/customer-topic-0/.log --print-data-log
> >
> >
> >
> > If I use it, I get the same output as:
> >
> > ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> > logs/kafka-0/customer-topic-0/.log
> >
> > These previous arguments, aren't take account ?
> >
> >
> > best regards,
> >
> >
> > Adrien
> >
> >
>

Re: DumpLogSegment

[Ted Yu]

For --index-sanity-check, according to dumpIndex():

if (indexSanityOnly) {

  index.sanityCheck

  println(s"$file passed sanity check.")


Do you see the print above ?


Which release of Kafka are you using ?


Cheers


On Sat, Feb 10, 2018 at 1:54 PM, adrien ruffie 
wrote:

> Hi all,
>
> In Kafka the definitive guide in page 200-201, two parameters of
> kafka.tools.DumpLogSegments appear not really work ...
>
> the --index-sanity-check argument
>
> the --print-data-log
>
> exemple:
>
> ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> logs/kafka-0/customer-topic-0/.log
> --index-sanity-check
>
> ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> logs/kafka-0/customer-topic-0/.log --print-data-log
>
>
>
> If I use it, I get the same output as:
>
> ./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
> logs/kafka-0/customer-topic-0/.log
>
> These previous arguments, aren't take account ?
>
>
> best regards,
>
>
> Adrien
>
>

Re: question on serialization ..

[Ted Yu]

Please read the javadoc:
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/Consumed.java

and correlate with the sample code.

Thanks

On Sat, Feb 10, 2018 at 1:10 PM, Debasish Ghosh <ghosh.debas...@gmail.com>
wrote:

> Looking at
> https://github.com/confluentinc/kafka-streams-
> examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> StreamToTableJoinScalaIntegrationTest.scala#L148,
> it seems that the leftJoin generates a KStream[String, (String, Long)],
> which means the value is a tuple of (String, Long) .. I am not able to get
> how this will serialize/de-serialize with the default serializers which are
> both stringSerde for keys and values.
>
> or am I missing something ?
>
> regards.
>
> On Sun, Feb 11, 2018 at 2:30 AM, Ted Yu <yuzhih...@gmail.com> wrote:
>
> > If I read the code correctly, the operation on this line prepares the
> input
> > for the (stringSerde, stringSerde) specified on line 142:
> >
> >   .leftJoin(userRegionsTable, (clicks: Long, region: String) => (if
> > (region == null) "UNKNOWN" else region, clicks))
> >
> > FYI
> >
> > On Sat, Feb 10, 2018 at 11:00 AM, Debasish Ghosh <
> ghosh.debas...@gmail.com
> > >
> > wrote:
> >
> > > Hi -
> > >
> > > I was going through this example at
> > > https://github.com/confluentinc/kafka-streams-
> > > examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> > > StreamToTableJoinScalaIntegrationTest.scala,
> > > especially the leftJoin part
> > > https://github.com/confluentinc/kafka-streams-
> > > examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> > > StreamToTableJoinScalaIntegrationTest.scala#L156.
> > > This leftJoin returns KStream[String, (String, Long)], while default
> > > serializers are String for both key and value as in
> > > https://github.com/confluentinc/kafka-streams-
> > > examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> > > StreamToTableJoinScalaIntegrationTest.scala#L112-L113.
> > > My question is how does this serialization work here ? I mean how does
> > the
> > > tuple get serialized with the default serializers ? And leftJoin only
> > works
> > > with default serializers ..
> > >
> > > regards.
> > >
> > > --
> > > Debasish Ghosh
> > > http://manning.com/ghosh2
> > > http://manning.com/ghosh
> > >
> > > Twttr: @debasishg
> > > Blog: http://debasishg.blogspot.com
> > > Code: http://github.com/debasishg
> > >
> >
>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>

Re: question on serialization ..

[Ted Yu]

If I read the code correctly, the operation on this line prepares the input
for the (stringSerde, stringSerde) specified on line 142:

  .leftJoin(userRegionsTable, (clicks: Long, region: String) => (if
(region == null) "UNKNOWN" else region, clicks))

FYI

On Sat, Feb 10, 2018 at 11:00 AM, Debasish Ghosh 
wrote:

> Hi -
>
> I was going through this example at
> https://github.com/confluentinc/kafka-streams-
> examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> StreamToTableJoinScalaIntegrationTest.scala,
> especially the leftJoin part
> https://github.com/confluentinc/kafka-streams-
> examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> StreamToTableJoinScalaIntegrationTest.scala#L156.
> This leftJoin returns KStream[String, (String, Long)], while default
> serializers are String for both key and value as in
> https://github.com/confluentinc/kafka-streams-
> examples/blob/3.3.x/src/test/scala/io/confluent/examples/streams/
> StreamToTableJoinScalaIntegrationTest.scala#L112-L113.
> My question is how does this serialization work here ? I mean how does the
> tuple get serialized with the default serializers ? And leftJoin only works
> with default serializers ..
>
> regards.
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>

Re: Information regarding Kafka 2.10-0.10.1.1 version

[Ted Yu]

For #1, there is record-size-avg metric

Not sure about #2

On Thu, Feb 8, 2018 at 10:28 AM, Pawan K  wrote:

> Hi,
> I am currently trying to research answers for the following questions. Can
> you please let me know where/how could i find these in the configuration.
>
> 1) Average record size (KB) for data written to each kafka topic
> 2) Average number of events written to each kafka topic per day
>
>
> Thanks,
> Pavan
>

Re: Question Related to Zookeeper and kafka java error message.

[Ted Yu]

Are you using zookeeper 3.4.11 ?

Please take a look at ZOOKEEPER-2976

Can you pastebin more log before the NoRouteToHostException ?

Cheers

On Thu, Feb 8, 2018 at 11:39 AM, Raymond, Shawn P CTR USARMY NETCOM (US) <
shawn.p.raymond@mail.mil> wrote:

> Afternoon all,
>
> I was wondering if anyone has seen the following messages appearing after
> deployment of a quorum.
>
> I am running the latest versions of kafka and zookeeper available.
>
> kafka system 1
> error OpenJDK 64-Bit Server VM warning: If the number of processors is
> expected to increase from one, then you should configure the number of
> parallel GC threads appropriately using -XX:ParallelGCThreads=N
>
> kafka systems 2 and 3
> OpenJDK 64-Bit Server VM warning: If the number of processors is expected
> to increase from one, then you should configure the number of parallel GC
> threads appropriately using -XX:ParallelGCThreads=N
> [2018-02-08 12:36:45,348] WARN Session 0x0 for server null, unexpected
> error, closing socket connection and attempting reconnect
> (org.apache.zookeeper.ClientCnxn)
> java.net.NoRouteToHostException: No route to host
> at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
> at sun.nio.ch.SocketChannelImpl.finishConnect(
> SocketChannelImpl.java:717)
> at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(
> ClientCnxnSocketNIO.java:361)
> at org.apache.zookeeper.ClientCnxn$SendThread.run(
> ClientCnxn.java:1141)
>
> does anyone have a clue how to fix these errors, I am running on RHEL 7.4.
>
> the strange thing is replication is working and topics are being listed.
>
> Any assistance would be appreciated.
>
> thank you.
>
> V/R
>
> 
>
> Shawn P. Raymond I
>
> RHCSA, CISSP, GSEC, GISP, SCSA, MCSA, MCNPS,  MCPS, MCP
>
> NCI Inc., Subject Matter Expert Linux /Senior Systems Engineer
>
> NETCOM G-3/5/7 Shared Resources OED
>
> DSN: 520-879-5331
>
> TEL: 520-533-5331
>
>

Re: mistake in Kafka the definitive guide

[Ted Yu]

I think Paolo's comment is correct.

BTW the paragraph above the figure says

Kafka currently lacks transactions

The above seems to have been written some time ago.

Cheers

On Thu, Feb 8, 2018 at 2:10 PM, adrien ruffie 
wrote:

> I take a look the errata page, and I saw that Paolo had already entered
> the errata. But is unconfirmed for the moment
>
>
> PDF Page 166
> Figure 8-5
>
> The labels on the topics/partitions are all the same: "Topic A, Partition
> 0". I think they should be (from top to bottom): "Topic A, Partition 0"
> "Topic B, Partition 0" "Topic __conusmer_offsets"
> Paolo Baronti   Dec 01, 2017
>
>
> 
> De : adrien ruffie 
> Envoyé : jeudi 8 février 2018 22:14:53
> À : users@kafka.apache.org
> Objet : mistake in Kafka the definitive guide
>
> Hello all,
>
>
> I'm reading Kafka the definitive guide and I suspect that found an error
> in the page 166 figure "Figure 8-5 a fail over causes committed offsets
> without matching records".
>
>
> In the figure we can't see Topic B ... specified in the box "Group C1,
> Topic B, Parition 0, Offset 6" ... the figure is not correct?
>
>
> Thus, I don't really understand the figure ...
>
>
> best regards,
>
>
> Adrien
>

Re: Cancel partition reassignment?

[Ted Yu]

Have you seen this thread ?

http://search-hadoop.com/m/Kafka/uyzND1pHiNuYt8hc1?subj=Re+Question+Kafka+Reassign+partitions+tool

On Thu, Feb 8, 2018 at 4:12 PM, Dylan Martin 
wrote:

> Hi all.
>
>
> I'm trying to cancel a failed partition reassignment.  I've heard that
> this can be done by deleting /admin/reassign_partitions in zookeeper.  I've
> tried and /admin/reassign_partitions won't go away.
>
>
> Does anyone know a way to cancel a partition reassignment?
>
>
> -Dylan
>
>
> (206) 855-9740 - Home
>
> (206) 235-8809 - Cell
>
> The information contained in this email message, and any attachment
> thereto, is confidential and may not be disclosed without the sender's
> express permission. If you are not the intended recipient or an employee or
> agent responsible for delivering this message to the intended recipient,
> you are hereby notified that you have received this message in error and
> that any review, dissemination, distribution or copying of this message, or
> any attachment thereto, in whole or in part, is strictly prohibited. If you
> have received this message in error, please immediately notify the sender
> by telephone, fax or email and delete the message and all of its
> attachments. Thank you.
>

Re: [VOTE] 1.0.1 RC0

[Ted Yu]

+1

Checked signature
Ran test suite where there was one flaky test (KAFKA-5889):

kafka.metrics.MetricsTest > testMetricsLeak FAILED
java.lang.AssertionError: expected:<1365> but was:<1368>
at org.junit.Assert.fail(Assert.java:88)
at org.junit.Assert.failNotEquals(Assert.java:834)
at org.junit.Assert.assertEquals(Assert.java:645)
at org.junit.Assert.assertEquals(Assert.java:631)
at kafka.metrics.MetricsTest$$anonfun$testMetricsLeak$1.
apply$mcVI$sp(MetricsTest.scala:69)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)
at kafka.metrics.MetricsTest.testMetricsLeak(MetricsTest.scala:67)

On Mon, Feb 5, 2018 at 7:48 PM, Ewen Cheslack-Postava 
wrote:

> Hello Kafka users, developers and client-developers,
>
> Sorry for a bit of delay, but I've now prepared the first candidate for
> release of Apache Kafka 1.0.1.
>
> This is a bugfix release for the 1.0 branch that was first released with
> 1.0.0 about 3 months ago. We've fixed 46 significant issues since that
> release. Most of these are non-critical, but in aggregate these fixes will
> have significant impact. A few of the more significant fixes include:
>
> * KAFKA-6277: Make loadClass thread-safe for class loaders of Connect
> plugins
> * KAFKA-6185: Selector memory leak with high likelihood of OOM in case of
> down conversion
> * KAFKA-6269: KTable state restore fails after rebalance
> * KAFKA-6190: GlobalKTable never finishes restoring when consuming
> transactional messages
>
> Release notes for the 1.0.1 release:
> http://home.apache.org/~ewencp/kafka-1.0.1-rc0/RELEASE_NOTES.html
>
> *** Please download, test and vote by Thursday, Feb 8, 12pm PT ***
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~ewencp/kafka-1.0.1-rc0/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~ewencp/kafka-1.0.1-rc0/javadoc/
>
> * Tag to be voted upon (off 1.0 branch) is the 1.0.1 tag:
> https://github.com/apache/kafka/tree/1.0.1-rc0
>
>
> * Documentation:
> http://kafka.apache.org/10/documentation.html
>
> * Protocol:
> http://kafka.apache.org/10/protocol.html
>
>
> Please test and verify the release artifacts and submit a vote for this RC,
> or report any issues so we can fix them and get a new RC out ASAP! Although
> this release vote requires PMC votes to pass, testing, votes, and bug
> reports are valuable and appreciated from everyone.
>
> Thanks,
> Ewen
>

Re: log.retention.bytes not working as expected

[Ted Yu]

The log cleaner abortion in the log file preceded log deletion.

On Tue, Feb 6, 2018 at 1:36 PM, Raghav <raghavas...@gmail.com> wrote:

> Ted
>
> Sorry, I did not understand your point here.
>
> On Tue, Feb 6, 2018 at 1:09 PM, Ted Yu <yuzhih...@gmail.com> wrote:
>
> > bq. but is aborted.
> >
> > See the following in LogManager#asyncDelete():
> >
> >   //We need to wait until there is no more cleaning task on the log
> to
> > be deleted before actually deleting it.
> >
> >   if (cleaner != null && !isFuture) {
> >
> > cleaner.abortCleaning(topicPartition)
> >
> > FYI
> >
> > On Tue, Feb 6, 2018 at 12:56 PM, Raghav <raghavas...@gmail.com> wrote:
> >
> > > From the log-cleaner.log, I see the following. It seems like it resume
> > but
> > > is aborted. Not sure how to read this:
> > >
> > >
> > > [2018-02-06 18:06:22,178] INFO Compaction for partition topic043-27 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,178] INFO The cleaning for partition topic043-27
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,182] INFO The cleaning for partition topic043-51
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,182] INFO Compaction for partition topic043-51 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,182] INFO The cleaning for partition topic043-51
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,190] INFO The cleaning for partition topic043-52
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,190] INFO Compaction for partition topic043-52 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,190] INFO The cleaning for partition topic043-52
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,192] INFO The cleaning for partition topic043-45
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,192] INFO Compaction for partition topic043-45 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,192] INFO The cleaning for partition topic043-45
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,198] INFO The cleaning for partition topic043-20
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,198] INFO Compaction for partition topic043-20 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,198] INFO The cleaning for partition topic043-20
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,204] INFO The cleaning for partition topic043-63
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,204] INFO Compaction for partition topic043-63 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,204] INFO The cleaning for partition topic043-63
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,209] INFO The cleaning for partition topic043-44
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,209] INFO Compaction for partition topic043-44 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,209] INFO The cleaning for partition topic043-44
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,214] INFO The cleaning for partition topic043-38
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,214] INFO Compaction for partition topic043-38 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,214] INFO The cleaning for partition topic043-38
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,219] INFO The cleaning for partition topic043-50
> is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,219] INFO Compaction for partition topic043-50 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,219] INFO The cleaning for partition topic043-50
> is
> > > aborted (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,224] INFO The cleaning for partition topic043-2 is
> > > aborted and paused (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,224] INFO Compaction for partition topic043-2 is
> > > resumed (kafka.log.LogCleaner)
> > > [2018-02-06 18:06:22,224] INFO The cleaning for partition to

Re: log.retention.bytes not working as expected

[Ted Yu]

bq. but is aborted.

See the following in LogManager#asyncDelete():

  //We need to wait until there is no more cleaning task on the log to
be deleted before actually deleting it.

  if (cleaner != null && !isFuture) {

cleaner.abortCleaning(topicPartition)

FYI

On Tue, Feb 6, 2018 at 12:56 PM, Raghav  wrote:

> From the log-cleaner.log, I see the following. It seems like it resume but
> is aborted. Not sure how to read this:
>
>
> [2018-02-06 18:06:22,178] INFO Compaction for partition topic043-27 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,178] INFO The cleaning for partition topic043-27 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,182] INFO The cleaning for partition topic043-51 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,182] INFO Compaction for partition topic043-51 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,182] INFO The cleaning for partition topic043-51 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,190] INFO The cleaning for partition topic043-52 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,190] INFO Compaction for partition topic043-52 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,190] INFO The cleaning for partition topic043-52 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,192] INFO The cleaning for partition topic043-45 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,192] INFO Compaction for partition topic043-45 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,192] INFO The cleaning for partition topic043-45 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,198] INFO The cleaning for partition topic043-20 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,198] INFO Compaction for partition topic043-20 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,198] INFO The cleaning for partition topic043-20 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,204] INFO The cleaning for partition topic043-63 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,204] INFO Compaction for partition topic043-63 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,204] INFO The cleaning for partition topic043-63 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,209] INFO The cleaning for partition topic043-44 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,209] INFO Compaction for partition topic043-44 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,209] INFO The cleaning for partition topic043-44 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,214] INFO The cleaning for partition topic043-38 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,214] INFO Compaction for partition topic043-38 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,214] INFO The cleaning for partition topic043-38 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,219] INFO The cleaning for partition topic043-50 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,219] INFO Compaction for partition topic043-50 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,219] INFO The cleaning for partition topic043-50 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,224] INFO The cleaning for partition topic043-2 is
> aborted and paused (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,224] INFO Compaction for partition topic043-2 is
> resumed (kafka.log.LogCleaner)
> [2018-02-06 18:06:22,224] INFO The cleaning for partition topic043-2 is
> aborted (kafka.log.LogCleaner)
> [2018-02-06 18:06:54,643] INFO Shutting down the log cleaner.
> (kafka.log.LogCleaner)
> [2018-02-06 18:06:54,643] INFO [kafka-log-cleaner-thread-0], Shutting down
> (kafka.log.LogCleaner)
> [2018-02-06 18:06:54,644] INFO [kafka-log-cleaner-thread-0], Shutdown
> completed (kafka.log.LogCleaner)
> [2018-02-06 18:06:54,644] INFO [kafka-log-cleaner-thread-0], Stopped
>  (kafka.log.LogCleaner)
> [2018-02-06 18:06:57,663] INFO Starting the log cleaner
> (kafka.log.LogCleaner)
> [2018-02-06 18:06:57,665] INFO [kafka-log-cleaner-thread-0], Starting
>  (kafka.log.LogCleaner)
> [2018-02-06 18:08:07,187] INFO Shutting down the log cleaner.
> (kafka.log.LogCleaner)
> [2018-02-06 18:08:07,187] INFO [kafka-log-cleaner-thread-0], Shutting down
> (kafka.log.LogCleaner)
> [2018-02-06 18:08:07,187] INFO [kafka-log-cleaner-thread-0], Stopped
>  (kafka.log.LogCleaner)
> [2018-02-06 18:08:07,187] INFO [kafka-log-cleaner-thread-0], Shutdown
> completed (kafka.log.LogCleaner)
> [2018-02-06 18:08:11,701] INFO Starting the log cleaner
> (kafka.log.LogCleaner)
> [2018-02-06 18:08:11,703] INFO [kafka-log-cleaner-thread-0], Starting
>  (kafka.log.LogCleaner)
>

Re: Strange Topic ...

[Ted Yu]

Which Kafka version are you using ?
Older versions of kafka (0.10 and prior) had some bugs in the log-cleaner
thread that might sometimes cause it to crash.

Please check the log-cleaner.log file to see if there was some clue.

Cheers

On Sun, Feb 4, 2018 at 11:14 AM, adrien ruffie 
wrote:

> Hello all,
>
>
> I'm a beginner in Kafka and this morning when I try some tests and when
> running this following cmd:
>
> ./bin kafka-topics.sh --zookeeper localhost:2181 --describe
>
>
> I understand my 3 created topic like "customer-topic",
> "streams-plaintext-input", and "streams-wordcount-output"
>
>
> But I already get this following output, why __consumer_offsets have 50
> partitions ! I never created it ... do you know this beavior ?
>
>
> Topic:__consumer_offsetsPartitionCount:50
>  ReplicationFactor:1 Configs:segment.bytes=104857600,cleanup.policy=
> compact,compression.type=produ$
> Topic: __consumer_offsets   Partition: 0Leader: 0
>  Replicas: 0 Isr: 0
> Topic: __consumer_offsets   Partition: 1Leader: 0
>  Replicas: 0 Isr: 0
> Topic: __consumer_offsets   Partition: 2Leader: 0
>  Replicas: 0 Isr: 0
> Topic: __consumer_offsets   Partition: 3Leader: 0
>  Replicas: 0 Isr: 0
> Topic: __consumer_offsets   Partition: 4Leader: 0
>  Replicas: 0 Isr: 0
> Topic: __consumer_offsets   Partition: 5Leader: 0
>  Replicas: 0 Isr: 0
> 
> Topic: __consumer_offsets   Partition: 49   Leader: 0
>  Replicas: 0 Isr: 0
>
>
> Topic:customer-topicPartitionCount:1ReplicationFactor:1
>  Configs:
> Topic: customer-topic   Partition: 0Leader: 0   Replicas:
> 0 Isr: 0
> Topic:streams-plaintext-input   PartitionCount:1
> ReplicationFactor:1 Configs:
> Topic: streams-plaintext-input  Partition: 0Leader: 0
>  Replicas: 0 Isr: 0
> Topic:streams-wordcount-output  PartitionCount:1
> ReplicationFactor:1 Configs:cleanup.policy=compact
> Topic: streams-wordcount-output Partition: 0Leader: 0
>  Replicas: 0 Isr: 0
>
>
> Thank and bests regards,
>
> Adrien
>
>
>
>

Re: Recommended max number of topics (and data separation)

[Ted Yu]

After brief search, I found KAFKA-6469
FYI
 Original message From: Andrey Falko  
Date: 2/1/18  5:28 PM  (GMT-08:00) To: users@kafka.apache.org Subject: Re: 
Recommended max number of topics (and data separation) 
Indeed David, I confirmed that I can't push my clusters to more than
72k topics with default zookeeper settings. Once I get to that
quantity, leader election never happens for new topics. Additionally
if I kill one of the brokers, all the topics don't get leaders
re-elected and it is impossible to trigger it due to the ClientCnxn
"Packet len4194494 is out of range" exception or "Failed to start
preferred replica election" error. It seems like there should be a
couple of JIRA item for starters:
1) Don't let # of topics exceed what jute.maxbuffer will bear; i.e.
send a good error message back to users saying the peak has been
reached and no more topics can be created.
2) Kafka should support more than 72k topics (probably without
requiring messing with jute.maxbuffer).

Is anyone aware of JIRA tickets that might already cover the above?

On Wed, Jan 31, 2018 at 8:35 AM, David Espinosa  wrote:
> I used:
> -Djute.maxbuffer=50111000
> and the gain I had is that I could increment number of topics from 70k to
> 100k :P
>
> 2018-01-30 23:25 GMT+01:00 Andrey Falko :
>
>> On Tue, Jan 30, 2018 at 1:38 PM, David Espinosa  wrote:
>> > Hi Andrey,
>> > My topics are replicated with a replicated factor equals to the number of
>> > nodes, 3 in this test.
>> > Didn't know about the kip-227.
>> > The problems I see at 70k topics coming from ZK are related to any
>> > operation where ZK has to retrieve topics metadata. Just listing topics
>> at
>> > 50K or 60k you will experience a big delay in the response. I have no
>> more
>> > details about these problems, but is easy to reproduce the latency in the
>> > topics list request.
>>
>> AFAIK kafka doesn't do a full list as part of normal operations from
>> ZK. If you have requirements in your consumer/producer code on doing
>> --describe, then that would be a problem. I think that can be worked
>> around. Based on my profiling data so far, while things are working in
>> non-failure mode, none of the ZK functions pop up as "hot methods".
>>
>> > Thanks me for pointing me to this parameter,  vm.max_map_count, it wasn't
>> > on my radar. Could you tell me what value you use?
>>
>> I set it to the max allowable on Amzn Linux: vm.max_map_count=1215752192
>>
>> > The other way around about topic naming, I think the longer the topic
>> names
>> > are the sooner jute.maxbuffer overflows.
>>
>> I see; what value(s) have you tried with and how much gain did you you see?
>>
>> > David
>> >
>> >
>> > 2018-01-30 4:40 GMT+01:00 Andrey Falko :
>> >
>> >> On Sun, Jan 28, 2018 at 8:45 AM, David Espinosa 
>> wrote:
>> >> > Hi Monty,
>> >> >
>> >> > I'm also planning to use a big amount of topics in Kafka, so recently
>> I
>> >> > made a test within a 3 nodes kafka cluster where I created 100k topics
>> >> with
>> >> > one partition. Sent 1M messages in total.
>> >>
>> >> Are your topic partitions replicated?
>> >>
>> >> > These are my conclusions:
>> >> >
>> >> >    - There is not any limitation on kafka regarding the number of
>> topics
>> >> >    but on Zookeeper and in the system where Kafka nodes is allocated.
>> >>
>> >> There are also the problems being addressed in KIP-227:
>> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>> >> 227%3A+Introduce+Incremental+FetchRequests+to+Increase+
>> >> Partition+Scalability
>> >>
>> >> >    - Zookeeper will start having problems from 70k topics, which can
>> be
>> >> >    solved modifying a buffer parameter on the JVM (-Djute.maxbuffer).
>> >> >    Performance is reduced.
>> >>
>> >> What kind of problems do you see at 70k topics? If performance is
>> >> reduced w/ modifying jute.maxbuffer, won't that effect the performance
>> >> of kafka interms of how long it takes to recover from broker failure,
>> >> creating/deleting topics, producing and consuming?
>> >>
>> >> >    - Open file descriptors of the system are equivalent to [number of
>> >> >    topics]X[number of partitions per topic]. Set to 128k in my test to
>> >> avoid
>> >> >    problems.
>> >> >    - System needs a big amount of memory for page caching.
>> >>
>> >> I also had to tune vm.max_map_count much higher.
>> >>
>> >> >
>> >> > So, after creating 100k with the required setup (system+JVM) but
>> seeing
>> >> > problems at 70k, I feel safe by not creating more than 50k, and always
>> >> will
>> >> > have Zookeeper as my first suspect if a problem comes. I think with
>> >> proper
>> >> > resources (memory) and system setup (open file descriptors), you don't
>> >> have
>> >> > any real limitation regarding partitions.
>> >>
>> >> I can confirm the 50k number. After about 40k-45k topics, I start
>> >> seeing slow down in consume 

Re: Memory Leak in Kafka

[Ted Yu]

Did you attach two .png files ?

Please use third party site since the attachment didn't come thru.

On Tue, Jan 23, 2018 at 5:20 PM, Avinash Herle 
wrote:

>
> Hi,
>
> I'm using Kafka as a messaging system in my data pipeline. I've a couple
> of producer processes in my pipeline and Spark Streaming
> 
> and Druid's Kafka indexing service
> 
> as consumers of Kafka. The indexing service spawns 40 new indexing tasks
> (Kafka consumers) every 15 mins.
>
> The heap memory used on Kafka seems fairly constant for an hour after
> which it seems to shoot up to the max allocated space. The garbage
> collection logs of Kafka seems to indicate a memory leak in Kafka. Find
> attached the plots generated from the GC logs.
>
> *Kafka Deployment:*
> 3 nodes, with 3 topics and 64 partitions per topic
>
> *Kafka Runtime jvm parameters:*
> 8GB Heap Memory
> 1GC swap Memory
> Using G1GC
> MaxGCPauseMilllis=20
> InitiatingHeapOccupancyPercent=35
>
> *Kafka Versions Used:*
> I've used Kafka version 0.10.0, 0.11.0.2 and 1.0.0 and find similar
> behavior
>
> *Questions:*
> 1) Is this a memory leak on the Kafka side or a misconfiguration of my
> Kafka cluster?
> 2) Druid creates new indexing tasks periodically. Does Kafka stably handle
> large number of consumers being added periodically?
> 3) As a knock on effect, We also notice kafka partitions going offline
> periodically after some time with the following error:
> ERROR [ReplicaFetcherThread-18-2], Error for partition [topic1,2] to
> broker 2:*org.apache.kafka.common.errors.UnknownTopicOrPartitionException*:
> This server does not host this topic-partition. (kafka.server.
> ReplicaFetcherThread)
>
> Can someone shed some light on the behavior being seen in my cluster?
>
> Please let me know if more details are needed to root cause the behavior
> being seen.
>
> Thanks in advance.
>
> Avinash
> [image: Screen Shot 2018-01-23 at 2.29.04 PM.png][image: Screen Shot
> 2018-01-23 at 2.29.21 PM.png]
>
>
>
>
> --
>
> Excuse brevity and typos. Sent from mobile device.
>
>
> --
>
> Excuse brevity and typos. Sent from mobile device.
>

Re: [VOTE] KIP-247: Add public test utils for Kafka Streams

[Ted Yu]

+1
 Original message From: Guozhang Wang  
Date: 1/17/18  10:38 PM  (GMT-08:00) To: users@kafka.apache.org Subject: Re: 
[VOTE] KIP-247: Add public test utils for Kafka Streams 
+1 (binding).

On Wed, Jan 17, 2018 at 6:09 PM, Matthias J. Sax 
wrote:

> Hi,
>
> I would like to start the vote for KIP-247:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 247%3A+Add+public+test+utils+for+Kafka+Streams
>
>
> -Matthias
>
>


-- 
-- Guozhang

Re: [ANNOUNCE] New committer: Matthias J. Sax

[Ted Yu]

Congratulations, Matthias.

On Fri, Jan 12, 2018 at 2:59 PM, Guozhang Wang  wrote:

> Hello everyone,
>
> The PMC of Apache Kafka is pleased to announce Matthias J. Sax as our
> newest Kafka committer.
>
> Matthias has made tremendous contributions to Kafka Streams API since early
> 2016. His footprint has been all over the places in Streams: in the past
> two years he has been the main driver on improving the join semantics
> inside Streams DSL, summarizing all their shortcomings and bridging the
> gaps; he has also been largely working on the exactly-once semantics of
> Streams by leveraging on the transaction messaging feature in 0.11.0. In
> addition, Matthias have been very active in community activity that goes
> beyond mailing list: he's getting the close to 1000 up votes and 100
> helpful flags on SO for answering almost all questions about Kafka Streams.
>
> Thank you for your contribution and welcome to Apache Kafka, Matthias!
>
>
>
> Guozhang, on behalf of the Apache Kafka PMC
>

Re: Broker won't exit...

[Ted Yu]

Skip:Can you pastebin the stack trace of the stuck broker ?
Thanks
 Original message From: Skip Montanaro 
 Date: 1/10/18  3:52 AM  (GMT-08:00) To: 
users@kafka.apache.org Subject: Re: Broker won't exit... 
Did you stop the broker before stoping zookeeper?


Yes. My stop script executes the server stop scripts in reverse order from
my start script. Should I have stuck in a couple second sleep between
stopping the brokers and stopping zookeeper?

I was actually running two brokers. The one my stop script stopped first
exited properly.

Skip

Re: Broker won't exit...

[Ted Yu]

I think that is the default signal.
>From the script:

SIGNAL=${SIGNAL:-TERM}

FYI

On Wed, Jan 10, 2018 at 2:35 AM, Sam Pegler 
wrote:

> Have you tried a normal kill (sigterm) against the java process?
>
> __
>
> Sam Pegler
>
> PRODUCTION ENGINEER
>
> T. +44(0) 07 562 867 486
>
> 
> 3-7 Herbal Hill / London / EC1R 5EJ
> www.infectiousmedia.com
>
> This email and any attachments are confidential and may also be privileged.
> If you
> are not the intended recipient, please notify the sender immediately, and
> do not
> disclose the contents to another person, use it for any purpose, or store,
> or copy
> the information in any medium. Please also destroy and delete the message
> from
> your computer.
>
>
> On 9 January 2018 at 22:44, Skip Montanaro 
> wrote:
>
> > I only discovered the kafka-server-stop.sh script a couple days ago. I
> > can't seem to make it do its thing (the corresponding zookeeper stop
> > script seems to work just fine). All consumers have been stopped. Lsof
> > still shows the Kafka broker process listening on its port. The last
> > connection left the CLOSE_WAIT state several minutes ago. Gstack shows
> > 169 threads, most in pthread_cond_wait(), a handful in other wait-like
> > functions (sem_wait, pthread_join, pthread_cond_timedwait, poll,
> > epoll_wait). I'm running 2.11-1.0.0 on a Red Hat 6 server.
> >
> > What does it take to get a broker to exit (short of kill -9)?
> >
> > Thx,
> >
> > Skip Montanaro
> >
>

Re: Insanely long recovery time with Kafka 0.11.0.2

[Ted Yu]

Ismael:
We're on the same page.

0.11.0.2 was released on 17 Nov 2017.

By 'recently' in my previous email I meant the change was newer.

Vincent:
Did the machine your broker ran on experience power issue ?

Cheers

On Sat, Jan 6, 2018 at 7:36 AM, Ismael Juma <ism...@juma.me.uk> wrote:

> Hi Ted,
>
> The change you mention is not part of 0.11.0.2.
>
> Ismael
>
> On Sat, Jan 6, 2018 at 3:31 PM, Ted Yu <yuzhih...@gmail.com> wrote:
>
> > bq. WARN Found a corrupted index file due to requirement failed: Corrupt
> > index found, index file
> > (/data/kafka/data-processed-15/54942918.index)
> >
> > Can you search backward for 54942918.index in the log to see
> if
> > we can find the cause for corruption ?
> >
> > This part of code was recently changed by :
> >
> > KAFKA-6324; Change LogSegment.delete to deleteIfExists and harden log
> > recovery
> >
> > Cheers
> >
> > On Sat, Jan 6, 2018 at 7:18 AM, Vincent Rischmann <vinc...@rischmann.fr>
> > wrote:
> >
> > > Here's an excerpt just after the broker started:
> > > https://pastebin.com/tZqze4Ya
> > >
> > > After more than 8 hours of recovery the broker finally started. I
> haven't
> > > read through all 8 hours of log but the parts I looked at are like the
> > > pastebin.
> > >
> > > I'm not seeing much in the log cleaner logs either, they look normal.
> We
> > > have a couple of compacted topics but seems only the consumer offsets
> is
> > > ever compacted (the other topics don't have much traffic).
> > >
> > > On Sat, Jan 6, 2018, at 12:02 AM, Brett Rann wrote:
> > > > What do the broker logs say its doing during all that time?
> > > >
> > > > There are some consumer offset / log cleaner bugs which caused us
> > > similarly
> > > > log delays. that was easily visible by watching the log cleaner
> > activity
> > > in
> > > > the logs, and in our monitoring of partition sizes watching them go
> > down,
> > > > along with IO activity on the host for those files.
> > > >
> > > > On Sat, Jan 6, 2018 at 7:48 AM, Vincent Rischmann <
> > vinc...@rischmann.fr>
> > > > wrote:
> > > >
> > > > > Hello,
> > > > >
> > > > > so I'm upgrading my brokers from 0.10.1.1 to 0.11.0.2 to fix this
> bug
> > > > > https://issues.apache.org/jira/browse/KAFKA-4523
> > > > > <https://issues.apache.org/jira/browse/KAFKA-4523>
> > > > > Unfortunately while stopping one broker, it crashed exactly because
> > of
> > > > > this bug. No big deal usually, except after restarting Kafka in
> > > 0.11.0.2
> > > > > the recovery is taking a really long time.
> > > > > I have around 6TB of data on that broker, and before when it
> crashed
> > it
> > > > > usually took around 30 to 45 minutes to recover, but now I'm at
> > almost
> > > > > 5h since Kafka started and it's still not recovered.
> > > > > I'm wondering what could have changed to have such a dramatic
> effect
> > on
> > > > > recovery time ? Is there maybe something I can tweak to try to
> reduce
> > > > > the time ?
> > > > > Thanks.
> > > > >
> > >
> >
>

Re: Insanely long recovery time with Kafka 0.11.0.2

[Ted Yu]

bq. WARN Found a corrupted index file due to requirement failed: Corrupt
index found, index file
(/data/kafka/data-processed-15/54942918.index)

Can you search backward for 54942918.index in the log to see if
we can find the cause for corruption ?

This part of code was recently changed by :

KAFKA-6324; Change LogSegment.delete to deleteIfExists and harden log
recovery

Cheers

On Sat, Jan 6, 2018 at 7:18 AM, Vincent Rischmann 
wrote:

> Here's an excerpt just after the broker started:
> https://pastebin.com/tZqze4Ya
>
> After more than 8 hours of recovery the broker finally started. I haven't
> read through all 8 hours of log but the parts I looked at are like the
> pastebin.
>
> I'm not seeing much in the log cleaner logs either, they look normal. We
> have a couple of compacted topics but seems only the consumer offsets is
> ever compacted (the other topics don't have much traffic).
>
> On Sat, Jan 6, 2018, at 12:02 AM, Brett Rann wrote:
> > What do the broker logs say its doing during all that time?
> >
> > There are some consumer offset / log cleaner bugs which caused us
> similarly
> > log delays. that was easily visible by watching the log cleaner activity
> in
> > the logs, and in our monitoring of partition sizes watching them go down,
> > along with IO activity on the host for those files.
> >
> > On Sat, Jan 6, 2018 at 7:48 AM, Vincent Rischmann 
> > wrote:
> >
> > > Hello,
> > >
> > > so I'm upgrading my brokers from 0.10.1.1 to 0.11.0.2 to fix this bug
> > > https://issues.apache.org/jira/browse/KAFKA-4523
> > > 
> > > Unfortunately while stopping one broker, it crashed exactly because of
> > > this bug. No big deal usually, except after restarting Kafka in
> 0.11.0.2
> > > the recovery is taking a really long time.
> > > I have around 6TB of data on that broker, and before when it crashed it
> > > usually took around 30 to 45 minutes to recover, but now I'm at almost
> > > 5h since Kafka started and it's still not recovered.
> > > I'm wondering what could have changed to have such a dramatic effect on
> > > recovery time ? Is there maybe something I can tweak to try to reduce
> > > the time ?
> > > Thanks.
> > >
>

Re: Exception during topic deletion when Kafka is hosted in Docker in Windows.

[Ted Yu]

Which Kafka release are you using ?

Most likely /var/lib/kafka/test-0 was still being referenced by some thread.

There have been fixes in this area recently.

Cheers

On Fri, Jan 5, 2018 at 4:28 AM, Alex Galperin 
wrote:

> Hi,
> I host Kafka in Docker container in Windows. I mounted volume for storing
> Kafka data log.
> When I try to delete topic, I receive the following error:
>
> ERROR Error while deleting test-0 in dir /var/lib/kafka.
> (kafka.server.LogDirFailureChannel)
>  java.io.IOException: Failed to rename log directory from
> /var/lib/kafka/test-0 to
> /var/lib/kafka/test-0.a81ff9700e4e4c3e8b20c6d949971b64-delete
>  at kafka.log.LogManager.asyncDelete(LogManager.scala:671)
>  at kafka.cluster.Partition.$anonfun$delete$1(Partition.scala:178)
>  at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:217)
>  at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:225)
>  at kafka.cluster.Partition.delete(Partition.scala:173)
>  at kafka.server.ReplicaManager.stopReplica(ReplicaManager.scala:341)
>  at
> kafka.server.ReplicaManager.$anonfun$stopReplicas$2(
> ReplicaManager.scala:373)
>  at scala.collection.Iterator.foreach(Iterator.scala:929)
>  at scala.collection.Iterator.foreach$(Iterator.scala:929)
>  at scala.collection.AbstractIterator.foreach(Iterator.scala:1417)
>  at scala.collection.IterableLike.foreach(IterableLike.scala:71)
>  at scala.collection.IterableLike.foreach$(IterableLike.scala:70)
>  at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
>  at kafka.server.ReplicaManager.stopReplicas(ReplicaManager.scala:371)
>  at kafka.server.KafkaApis.handleStopReplicaRequest(KafkaApis.scala:190)
>  at kafka.server.KafkaApis.handle(KafkaApis.scala:104)
>  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:65)
>  at java.lang.Thread.run(Thread.java:748)
>
> Could somebody help me to cope with this issue?
>
> Thank you in advance,
> /Alex
>

Re: Streams - State store directory created automatically

[Ted Yu]

Looks like the .checkpoint file was generated from this code in
ProcessorStateManager
:

// write the checkpoint file before closing, to indicate clean
shutdown

try {

if (checkpoint == null) {

checkpoint = new OffsetCheckpoint(new File(baseDir,
CHECKPOINT_FILE_NAME));


FYI

On Thu, Jan 4, 2018 at 11:48 AM, Kristopher Kane 
wrote:

> I just noticed /tmp/kafka-streams//0_[0,1]/{.checkpoint,.lock]
> (there are two partitions on the incoming topic) being automatically
> created during an integration test.  My Streams app doesn't use a state
> store and only contains mapValues and a .to termination operation.
>
> Anyone know what this is for?
>
> Thanks,
>
> Kris
>

Re: Any suggestion about Kafka/Zookeeper migration.

[Ted Yu]

Did you intend to attach pictures following the two solutions ?

It seems the pictures didn't come thru.

FYI

On Wed, Jan 3, 2018 at 8:39 PM, Tony Liu  wrote:

> Hi All,
>
> This post here is aimed to ask experience about what did you do migration
> `Kafka/zookeeper` ?  :)
>
> All of Kafka/zookeeper are running on AWS, because of some reason, we have
> to replace all the existed server (you can simply think we will terminate
> the old server and create new server to replace it), which is running with
> Kafka/zookeeper.
>
> the Kafka infrastructure :
>
>- 3 zookeeper with exhibitor running together.
>- 6 brokers.
>
> requirements:
>
>- terminate all of the 3 zookeeper with exhibitor, and also 6 brokers.
>- launch 5 new zookeepers without exhibitor.
>- launch 6 new brokers.
>- 0 or minimum downtime, try our best no impact for the Kafka client.
>
>
> so far, I have two prepared solution, but any of them has pros & cons, the
> solution A is too much manual operations needed, the solution B is some
> development effort required and also need to migration all of the Kafka
> client to use the self-developed Kafka client with `fail-over` feature.
>
> *Solution A*: simply speaking, this solution is mainly working on manual
> operation to add znode / broker to the existed cluster, when all of the
> thing ready, I will step by step to terminate the old one.
>
>
> ​
>
> *Solution B:*  simply speaking, the solution will work like fail-over
> way, a new cluster will be launched, and control Kafka client's traffic to
> the new cluster, and terminate the old cluster.
>
>
> ​
>
>
>

Re: Unable to start 1st broker in a 3 node configuration

[Ted Yu]

bq.  zookeeper.connect = localhost:2181,eg2-pp-ifs-245:
2181,eg2-pp-ifs-219:*9092*

Why did 9092 appear in zookeeper setting ?

Cheers

On Tue, Jan 2, 2018 at 2:18 AM, M. Manna  wrote:

> Hi All,
>
> Firstly a very Happy New Year!
>
> I set up my 3 node configuration where each of the broker is set to have
> identical configurations. They are in in three different servers, but
> within the same domain.
>
> I have got a very simply Windows script that does the following:
>
> 1) Starts each zookeeper instances with a 5 seconds delay.
> 2) Once the zookeepers are running, wait for 10 seconds.
> 3) Start all Brokers with a 5 seconds delay.
>
> Just from today, I am unable to start my first broker. I tried to
> individually stop and start the broker but it didn't help. Also, I tried to
> do a full cleanpup (i.e. remove all ZK and Kafka logs) to start again. But
> the first broker seem to be causing issues.
>
> All these are in Kafka_2.10-0.10.2.1.
>
> I suspect that the machine has somehow bound 9092 port to something and got
> stuck in the process. At least, from the logs this is what I am getting:
>
> >
> > log4j:ERROR Failed to rename [C:\kafka_2.10-0.10.2.1/logs/server.log] to
> > [C:\kafka_2.10-0.10.2.1/logs/server.log.2018-01-02-09].
> > [2018-01-02 10:04:28,204] INFO KafkaConfig values:
> > advertised.host.name = null
> > advertised.listeners = PLAINTEXT://localhost:9092
> > advertised.port = null
> > authorizer.class.name =
> > auto.create.topics.enable = true
> > auto.leader.rebalance.enable = true
> > background.threads = 10
> > broker.id = 1
> > broker.id.generation.enable = true
> > broker.rack = null
> > compression.type = gzip
> > connections.max.idle.ms = 60
> > controlled.shutdown.enable = true
> > controlled.shutdown.max.retries = 10
> > controlled.shutdown.retry.backoff.ms = 3000
> > controller.socket.timeout.ms = 3
> > create.topic.policy.class.name = null
> > default.replication.factor = 1
> > delete.topic.enable = true
> > fetch.purgatory.purge.interval.requests = 1000
> > group.max.session.timeout.ms = 30
> > group.min.session.timeout.ms = 6000
> > host.name =
> > inter.broker.listener.name = null
> > inter.broker.protocol.version = 0.10.2-IV0
> > leader.imbalance.check.interval.seconds = 300
> > leader.imbalance.per.broker.percentage = 10
> > listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL
> > listeners = null
> > log.cleaner.backoff.ms = 15000
> > log.cleaner.dedupe.buffer.size = 134217728
> > log.cleaner.delete.retention.ms = 8640
> > log.cleaner.enable = true
> > log.cleaner.io.buffer.load.factor = 0.9
> > log.cleaner.io.buffer.size = 524288
> > log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
> > log.cleaner.min.cleanable.ratio = 0.5
> > log.cleaner.min.compaction.lag.ms = 0
> > log.cleaner.threads = 1
> > log.cleanup.policy = [delete]
> > log.dir = /tmp/kafka-logs
> > log.dirs = /kafka1
> > log.flush.interval.messages = 9223372036854775807
> > log.flush.interval.ms = null
> > log.flush.offset.checkpoint.interval.ms = 6
> > log.flush.scheduler.interval.ms = 9223372036854775807
> > log.index.interval.bytes = 4096
> > log.index.size.max.bytes = 10485760
> > log.message.format.version = 0.10.2-IV0
> > log.message.timestamp.difference.max.ms = 9223372036854775807
> > log.message.timestamp.type = CreateTime
> > log.preallocate = false
> > log.retention.bytes = 20971520
> > log.retention.check.interval.ms = 30
> > log.retention.hours = 2
> > log.retention.minutes = 15
> > log.retention.ms = null
> > log.roll.hours = 1
> > log.roll.jitter.hours = 0
> > log.roll.jitter.ms = null
> > log.roll.ms = null
> > log.segment.bytes = 10485760
> > log.segment.delete.delay.ms = 6
> > max.connections.per.ip = 2147483647
> > max.connections.per.ip.overrides =
> > message.max.bytes = 112
> > metric.reporters = []
> > metrics.num.samples = 2
> > metrics.recording.level = INFO
> > metrics.sample.window.ms = 3
> > min.insync.replicas = 1
> > num.io.threads = 24
> > num.network.threads = 9
> > num.partitions = 1
> > num.recovery.threads.per.data.dir = 1
> > num.replica.fetchers = 1
> > offset.metadata.max.bytes = 4096
> > offsets.commit.required.acks = -1
> > offsets.commit.timeout.ms = 5000
> > offsets.load.buffer.size = 5242880
> > offsets.retention.check.interval.ms = 30
> >  

Re: Partition reassignment data file is empty

[Ted Yu]

I logged KAFKA-6413 for improving error message
w.r.t. ReassignPartitionsCommand#parsePartitionReassignmentData()

FYI

On Sun, Dec 31, 2017 at 10:24 PM, allen chan <allen.michael.c...@gmail.com>
wrote:

> Absolutely user error. Works after i removed the erroneous comma. Wish the
> error message was more obvious.
> Thanks Brett and Ted!
>
> On Sun, Dec 31, 2017 at 6:29 PM, Ted Yu <yuzhih...@gmail.com> wrote:
>
> > I verified that Brett said thru this code:
> >
> > val (partitionsToBeReassigned, replicaAssignment) =
> > ReassignPartitionsCommand.parsePartitionReassignmentData(
> >
> > "{\"version\":1,\"partitions\":[{\"topic\":\"metrics\",\"
> > partition\"
> > :0,\"replicas\":[1,2]},{\"topic\":\"metrics\",\"
> partition\":1,\"replicas\"
> > :[2,3]},]}");
> >
> >
> > partitionsToBeReassigned was empty.
> >
> > I think parsePartitionReassignmentData() should be improved to give
> better
> > error information.
> >
> >
> > FYI
> >
> > On Sun, Dec 31, 2017 at 4:51 PM, Brett Rann <br...@zendesk.com.invalid>
> > wrote:
> >
> > > That's happening because your JSON is malformed. Losing the last comma
> > will
> > > fix it.
> > >
> > > On Sun, Dec 31, 2017 at 3:43 PM, allen chan <
> > allen.michael.c...@gmail.com>
> > > wrote:
> > >
> > > > Hello
> > > >
> > > > Kafka Version: 0.11.0.1
> > > >
> > > > I am trying to increase replication factor for a topic and i am
> getting
> > > the
> > > > below error. Can anyone help explain what the error means? The json
> is
> > > not
> > > > empty
> > > >
> > > > $ cat increase-replication-factor.json
> > > > {"version":1,
> > > >   "partitions":[
> > > >  {"topic":"metrics","partition":0,"replicas":[1,2]},
> > > >  {"topic":"metrics","partition":1,"replicas":[2,3]},
> > > > ]}
> > > >
> > > > $ sudo /opt/kafka/kafka_2.12-0.11.0.1/bin/kafka-reassign-
> partitions.sh
> > > > --zookeeper server1:2181 --reassignment-json-file
> > > > increase-replication-factor.json --execute
> > > > Partitions reassignment failed due to Partition reassignment data
> file
> > is
> > > > empty
> > > > kafka.common.AdminCommandFailedException: Partition reassignment
> data
> > > file
> > > > is empty
> > > > at
> > > > kafka.admin.ReassignPartitionsCommand$.parseAndValidate(
> > > > ReassignPartitionsCommand.scala:188)
> > > > at
> > > > kafka.admin.ReassignPartitionsCommand$.executeAssignment(
> > > > ReassignPartitionsCommand.scala:158)
> > > > at
> > > > kafka.admin.ReassignPartitionsCommand$.executeAssignment(
> > > > ReassignPartitionsCommand.scala:154)
> > > > at
> > > > kafka.admin.ReassignPartitionsCommand$.main(
> ReassignPartitionsCommand.
> > > > scala:51)
> > > > at
> > > > kafka.admin.ReassignPartitionsCommand.main(
> ReassignPartitionsCommand.
> > > > scala)
> > > >
> > > > Thanks
> > > > --
> > > > Allen Michael Chan
> > > >
> > >
> > >
> > >
> > > --
> > >
> > > Brett Rann
> > >
> > > Senior DevOps Engineer
> > >
> > >
> > > Zendesk International Ltd
> > >
> > > 395 Collins Street, Melbourne VIC 3000 Australia
> > >
> > > Mobile: +61 (0) 418 826 017
> > >
> >
>
>
>
> --
> Allen Michael Chan
>

Re: Partition reassignment data file is empty

[Ted Yu]

I verified that Brett said thru this code:

val (partitionsToBeReassigned, replicaAssignment) =
ReassignPartitionsCommand.parsePartitionReassignmentData(

"{\"version\":1,\"partitions\":[{\"topic\":\"metrics\",\"partition\"
:0,\"replicas\":[1,2]},{\"topic\":\"metrics\",\"partition\":1,\"replicas\"
:[2,3]},]}");


partitionsToBeReassigned was empty.

I think parsePartitionReassignmentData() should be improved to give better
error information.


FYI

On Sun, Dec 31, 2017 at 4:51 PM, Brett Rann 
wrote:

> That's happening because your JSON is malformed. Losing the last comma will
> fix it.
>
> On Sun, Dec 31, 2017 at 3:43 PM, allen chan 
> wrote:
>
> > Hello
> >
> > Kafka Version: 0.11.0.1
> >
> > I am trying to increase replication factor for a topic and i am getting
> the
> > below error. Can anyone help explain what the error means? The json is
> not
> > empty
> >
> > $ cat increase-replication-factor.json
> > {"version":1,
> >   "partitions":[
> >  {"topic":"metrics","partition":0,"replicas":[1,2]},
> >  {"topic":"metrics","partition":1,"replicas":[2,3]},
> > ]}
> >
> > $ sudo /opt/kafka/kafka_2.12-0.11.0.1/bin/kafka-reassign-partitions.sh
> > --zookeeper server1:2181 --reassignment-json-file
> > increase-replication-factor.json --execute
> > Partitions reassignment failed due to Partition reassignment data file is
> > empty
> > kafka.common.AdminCommandFailedException: Partition reassignment data
> file
> > is empty
> > at
> > kafka.admin.ReassignPartitionsCommand$.parseAndValidate(
> > ReassignPartitionsCommand.scala:188)
> > at
> > kafka.admin.ReassignPartitionsCommand$.executeAssignment(
> > ReassignPartitionsCommand.scala:158)
> > at
> > kafka.admin.ReassignPartitionsCommand$.executeAssignment(
> > ReassignPartitionsCommand.scala:154)
> > at
> > kafka.admin.ReassignPartitionsCommand$.main(ReassignPartitionsCommand.
> > scala:51)
> > at
> > kafka.admin.ReassignPartitionsCommand.main(ReassignPartitionsCommand.
> > scala)
> >
> > Thanks
> > --
> > Allen Michael Chan
> >
>
>
>
> --
>
> Brett Rann
>
> Senior DevOps Engineer
>
>
> Zendesk International Ltd
>
> 395 Collins Street, Melbourne VIC 3000 Australia
>
> Mobile: +61 (0) 418 826 017
>

Re: Async version of KafkaConsumer.partitionsFor()

[Ted Yu]

For #1, fetcher.getTopicMetadata() is called.
If you have time, you can read getTopicMetadata(). It is a blocking call
with given timeout.

For #2, I don't see any mechanism for metadata sharing.

FYI

On Fri, Dec 29, 2017 at 8:25 AM, Viliam Ďurina 
wrote:

> Hi,
>
> I use KafkaConsumer.partitionsFor() method to discover partitions that
> might be added at runtime. I use manual partition assignment. I call it
> once per second and rely on the metadata.max.age.ms property to throttle
> real number of remote calls.
>
> My questions:
>
>1.
>
>can the partitionsFor call block after we have initial metadata? I mean,
>when the metadata are expired, will it return latest metadata
> immediately
>and start fetching new metadata on the background or will it block.
>2.
>
>when I have multiple consumers for the same topic (one for each
>consuming thread), do they share metadata or each of them does a
> separate
>request.
>
> Thanks,
> Viliam
> ​
>