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

[Neha Narkhede]

Congratulations Matthias. Very well deserved!
On Fri, Jan 12, 2018 at 7:11 PM Manikumar  wrote:

> Congrats.  well deserved.
>
> On Sat, Jan 13, 2018 at 8:37 AM, Martin Gainty 
> wrote:
>
> > Willkommen Matthias!
> > Martin-
> > 
> > From: Damian Guy 
> > Sent: Friday, January 12, 2018 7:43 PM
> > To: users@kafka.apache.org
> > Cc: dev
> > Subject: Re: [ANNOUNCE] New committer: Matthias J. Sax
> >
> > Can't think of anyone me deserving! Congratulations Matthias!
> > On Sat, 13 Jan 2018 at 00:17, Ismael Juma  wrote:
> >
> > > Congratulations Matthias!
> > >
> > > On 12 Jan 2018 10: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
> > > >
> > >
> >
>
-- 
Thanks,
Neha

Re: [ANNOUNCE] New committer: Damian Guy

[Neha Narkhede]

Well deserved. Congratulations Damian!

On Fri, Jun 9, 2017 at 1:34 PM Guozhang Wang  wrote:

> Hello all,
>
>
> The PMC of Apache Kafka is pleased to announce that we have invited Damian
> Guy as a committer to the project.
>
> Damian has made tremendous contributions to Kafka. He has not only
> contributed a lot into the Streams api, but have also been involved in many
> other areas like the producer and consumer clients, broker-side
> coordinators (group coordinator and the ongoing transaction coordinator).
> He has contributed more than 100 patches so far, and have been driving on 6
> KIP contributions.
>
> More importantly, Damian has been a very prolific reviewer on open PRs and
> has been actively participating on community activities such as email lists
> and slack overflow questions. Through his code contributions and reviews,
> Damian has demonstrated good judgement on system design and code qualities,
> especially on thorough unit test coverages. We believe he will make a great
> addition to the committers of the community.
>
>
> Thank you for your contributions, Damian!
>
>
> -- Guozhang, on behalf of the Apache Kafka PMC
>
-- 
Thanks,
Neha

Re: [VOTE] KIP-156 Add option "dry run" to Streams application reset tool

[Neha Narkhede]

+1

On Wed, May 10, 2017 at 12:32 PM Gwen Shapira  wrote:

> +1. Also not sure that adding a parameter to a CLI requires a KIP. It seems
> excessive.
>
>
> On Tue, May 9, 2017 at 7:57 PM Jay Kreps  wrote:
>
> > +1
> > On Tue, May 9, 2017 at 3:41 PM BigData dev 
> > wrote:
> >
> > > Hi, Everyone,
> > >
> > > Since this is a relatively simple change, I would like to start the
> > voting
> > > process for KIP-156: Add option "dry run" to Streams application reset
> > tool
> > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=69410150
> > >
> > >
> > > The vote will run for a minimum of 72 hours.
> > >
> > >
> > > Thanks,
> > >
> > > Bharat
> > >
> >
>
-- 
Thanks,
Neha

Re: [VOTE] 0.10.2.1 RC1

[Neha Narkhede]

+1 (binding). Verified signatures, ran quickstart and tests.

On Fri, Apr 14, 2017 at 3:45 PM Gwen Shapira  wrote:

> Verified my own signatures, ran quickstart and created few Connectors.
>
> +1 (binding)
>
>
> On Wed, Apr 12, 2017 at 5:25 PM, Gwen Shapira  wrote:
> > Hello Kafka users, developers, client-developers, friends, romans,
> > citizens, etc,
> >
> > This is the second candidate for release of Apache Kafka 0.10.2.1.
> >
> > This is a bug fix release and it includes fixes and improvements from 24
> JIRAs
> > (including a few critical bugs).
> >
> > Release notes for the 0.10.2.1 release:
> > http://home.apache.org/~gwenshap/kafka-0.10.2.1-rc1/RELEASE_NOTES.html
> >
> > *** Please download, test and vote by Monday, April 17, 5:30 pm PT
> >
> > Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > Your help in validating this bugfix release is super valuable, so
> > please take the time to test and vote!
> >
> > Suggested tests:
> >  * Grab the source archive and make sure it compiles
> >  * Grab one of the binary distros and run the quickstarts against them
> >  * Extract and verify one of the site docs jars
> >  * Build a sample against jars in the staging repo
> >  * Validate GPG signatures on at least one file
> >  * Validate the javadocs look ok
> >  * The 0.10.2 documentation was updated for this bugfix release
> > (especially upgrade, streams and connect portions) - please make sure
> > it looks ok: http://kafka.apache.org/documentation.html
> >
> > * Release artifacts to be voted upon (source and binary):
> > http://home.apache.org/~gwenshap/kafka-0.10.2.1-rc1/
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Javadoc:
> > http://home.apache.org/~gwenshap/kafka-0.10.2.1-rc1/javadoc/
> >
> > * Tag to be voted upon (off 0.10.2 branch) is the 0.10.2.1 tag:
> >
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=e133f2ca57670e77f8114cc72dbc2f91a48e3a3b
> >
> > * Documentation:
> > http://kafka.apache.org/0102/documentation.html
> >
> > * Protocol:
> > http://kafka.apache.org/0102/protocol.html
> >
> > /**
> >
> > Thanks,
> >
> > Gwen Shapira
>
>
>
> --
> Gwen Shapira
> Product Manager | Confluent
> 650.450.2760 <(650)%20450-2760> | @gwenshap
> Follow us: Twitter | blog
>
-- 
Thanks,
Neha

Re: [ANNOUNCE] New committer: Grant Henke

[Neha Narkhede]

Congratulations, Grant. Well deserved!

On Thu, Jan 12, 2017 at 7:51 AM Grant Henke  wrote:

> Thanks everyone!
>
> On Thu, Jan 12, 2017 at 2:58 AM, Damian Guy  wrote:
>
> > Congratulations!
> >
> > On Thu, 12 Jan 2017 at 03:35 Jun Rao  wrote:
> >
> > > Grant,
> > >
> > > Thanks for all your contribution! Congratulations!
> > >
> > > Jun
> > >
> > > On Wed, Jan 11, 2017 at 2:51 PM, Gwen Shapira 
> wrote:
> > >
> > > > The PMC for Apache Kafka has invited Grant Henke to join as a
> > > > committer and we are pleased to announce that he has accepted!
> > > >
> > > > Grant contributed 88 patches, 90 code reviews, countless great
> > > > comments on discussions, a much-needed cleanup to our protocol and
> the
> > > > on-going and critical work on the Admin protocol. Throughout this, he
> > > > displayed great technical judgment, high-quality work and willingness
> > > > to contribute where needed to make Apache Kafka awesome.
> > > >
> > > > Thank you for your contributions, Grant :)
> > > >
> > > > --
> > > > Gwen Shapira
> > > > Product Manager | Confluent
> > > > 650.450.2760 <(650)%20450-2760> <(650)%20450-2760> | @gwenshap
> > > > Follow us: Twitter | blog
> > > >
> > >
> >
>
>
>
> --
> Grant Henke
> Software Engineer | Cloudera
> gr...@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>
-- 
Thanks,
Neha

Re: [kafka-clients] [VOTE] 0.10.1.0 RC3

[Neha Narkhede]

+1 (binding)

Verified quick start and artifacts.
On Mon, Oct 17, 2016 at 10:39 PM Dana Powers  wrote:

> +1 -- passes kafka-python integration tests
>
> On Mon, Oct 17, 2016 at 1:28 PM, Jun Rao  wrote:
> > Thanks for preparing the release. Verified quick start on scala 2.11
> binary.
> > +1
> >
> > Jun
> >
> > On Fri, Oct 14, 2016 at 4:29 PM, Jason Gustafson 
> wrote:
> >>
> >> Hello Kafka users, developers and client-developers,
> >>
> >> One more RC for 0.10.1.0. We're hoping this is the final one so that we
> >> can meet the release target date of Oct. 17 (Monday). Please let me
> know as
> >> soon as possible if you find any major problems.
> >>
> >> Release plan:
> >> https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.10.1.
> >>
> >> Release notes for the 0.10.1.0 release:
> >> http://home.apache.org/~jgus/kafka-0.10.1.0-rc3/RELEASE_NOTES.html
> >>
> >> *** Please download, test and vote by Monday, Oct 17, 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/~jgus/kafka-0.10.1.0-rc3/
> >>
> >> * Maven artifacts to be voted upon:
> >> https://repository.apache.org/content/groups/staging/
> >>
> >> * Javadoc:
> >> http://home.apache.org/~jgus/kafka-0.10.1.0-rc3/javadoc/
> >>
> >> * Tag to be voted upon (off 0.10.1 branch) is the 0.10.1.0-rc3 tag:
> >>
> >>
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=50f30a44f31fca1bd9189d2814388d51bd56b06b
> >>
> >> * Documentation:
> >> http://kafka.apache.org/0101/documentation.html
> >>
> >> * Protocol:
> >> http://kafka.apache.org/0101/protocol.html
> >>
> >> * Tests:
> >> Unit tests: https://builds.apache.org/job/kafka-0.10.1-jdk7/71/
> >> System tests:
> >>
> http://testing.confluent.io/confluent-kafka-0-10-1-system-test-results/?prefix=2016-10-13--001.1476369986--apache--0.10.1--ee212d1/
> >>
> >> (Note that these tests do not include a couple patches merged today. I
> >> will send links to updated test builds as soon as they are available)
> >>
> >> Thanks,
> >>
> >> Jason
> >>
> >> --
> >> You received this message because you are subscribed to the Google
> Groups
> >> "kafka-clients" group.
> >> To unsubscribe from this group and stop receiving emails from it, send
> an
> >> email to kafka-clients+unsubscr...@googlegroups.com.
> >> To post to this group, send email to kafka-clie...@googlegroups.com.
> >> Visit this group at https://groups.google.com/group/kafka-clients.
> >> To view this discussion on the web visit
> >>
> https://groups.google.com/d/msgid/kafka-clients/CAJDuW%3DBm0HCOjiHiwnW3WJ_i5u_0e4J2G_mZ_KBkB_WEmo7pNg%40mail.gmail.com
> .
> >> For more options, visit https://groups.google.com/d/optout.
> >
> >
> > --
> > You received this message because you are subscribed to the Google Groups
> > "kafka-clients" group.
> > To unsubscribe from this group and stop receiving emails from it, send an
> > email to kafka-clients+unsubscr...@googlegroups.com.
> > To post to this group, send email to kafka-clie...@googlegroups.com.
> > Visit this group at https://groups.google.com/group/kafka-clients.
> > To view this discussion on the web visit
> >
> https://groups.google.com/d/msgid/kafka-clients/CAFc58G_Atqyc7-O13EGnRNibng5UPo-a_2h00N2%3D%3DMtWktm%3D1g%40mail.gmail.com
> .
> >
> > For more options, visit https://groups.google.com/d/optout.
>
-- 
Thanks,
Neha

[ANNOUNCE] New committer: Jason Gustafson

[Neha Narkhede]

The PMC for Apache Kafka has invited Jason Gustafson to join as a committer and
we are pleased to announce that he has accepted!

Jason has contributed numerous patches to a wide range of areas, notably
within the new consumer and the Kafka Connect layers. He has displayed
great taste and judgement which has been apparent through his involvement
across the board from mailing lists, JIRA, code reviews to contributing
features, bug fixes and code and documentation improvements.

Thank you for your contribution and welcome to Apache Kafka, Jason!
-- 
Thanks,
Neha

Re: [VOTE] 0.10.0.1 RC2

[Neha Narkhede]

+1 (binding)

On Fri, Aug 5, 2016 at 12:29 PM, Grant Henke  wrote:

> +1 (non-binding)
>
> On Fri, Aug 5, 2016 at 2:04 PM, Dana Powers  wrote:
>
> > passed kafka-python integration tests, +1
> >
> > -Dana
> >
> >
> > On Fri, Aug 5, 2016 at 9:35 AM, Tom Crayford 
> wrote:
> > > Heroku has tested this using the same performance testing setup we used
> > to
> > > evaluate the impact of 0.9 -> 0.10 (see https://engineering.
> > > heroku.com/blogs/2016-05-27-apache-kafka-010-evaluating-
> > > performance-in-distributed-systems/).
> > >
> > > We see no issues at all with them, so +1 (non-binding) from here.
> > >
> > > On Fri, Aug 5, 2016 at 12:58 PM, Jim Jagielski 
> wrote:
> > >
> > >> Looks good here: +1
> > >>
> > >> > On Aug 4, 2016, at 9:54 AM, Ismael Juma  wrote:
> > >> >
> > >> > Hello Kafka users, developers and client-developers,
> > >> >
> > >> > This is the third candidate for the release of Apache Kafka
> 0.10.0.1.
> > >> This
> > >> > is a bug fix release and it includes fixes and improvements from 53
> > JIRAs
> > >> > (including a few critical bugs). See the release notes for more
> > details:
> > >> >
> > >> > http://home.apache.org/~ijuma/kafka-0.10.0.1-rc2/RELEASE_NOTES.html
> > >> >
> > >> > When compared to RC1, RC2 contains a fix for a regression where an
> > older
> > >> > version of slf4j-log4j12 was also being included in the libs folder
> of
> > >> the
> > >> > binary tarball (KAFKA-4008). Thanks to Manikumar Reddy for reporting
> > the
> > >> > issue.
> > >> >
> > >> > *** Please download, test and vote by Monday, 8 August, 8am 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/~ijuma/kafka-0.10.0.1-rc2/
> > >> >
> > >> > * Maven artifacts to be voted upon:
> > >> > https://repository.apache.org/content/groups/staging
> > >> >
> > >> > * Javadoc:
> > >> > http://home.apache.org/~ijuma/kafka-0.10.0.1-rc2/javadoc/
> > >> >
> > >> > * Tag to be voted upon (off 0.10.0 branch) is the 0.10.0.1-rc2 tag:
> > >> > https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=
> > >> f8f56751744ba8e55f90f5c4f3aed8c3459447b2
> > >> >
> > >> > * Documentation:
> > >> > http://kafka.apache.org/0100/documentation.html
> > >> >
> > >> > * Protocol:
> > >> > http://kafka.apache.org/0100/protocol.html
> > >> >
> > >> > * Successful Jenkins builds for the 0.10.0 branch:
> > >> > Unit/integration tests: *https://builds.apache.org/job
> > >> /kafka-0.10.0-jdk7/182/
> > >> > *
> > >> > System tests: *https://jenkins.confluent.io/
> > >> job/system-test-kafka-0.10.0/138/
> > >> > *
> > >> >
> > >> > Thanks,
> > >> > Ismael
> > >>
> > >>
> >
>
>
>
> --
> Grant Henke
> Software Engineer | Cloudera
> gr...@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>



-- 
Thanks,
Neha

[ANNOUNCE] New committer: Ismael Juma

[Neha Narkhede]

The PMC for Apache Kafka has invited Ismael Juma to join as a committer and
we are pleased to announce that he has accepted!

Ismael has contributed 121 commits
 to a wide range of
areas, notably within the security and the network layer. His involvement
has been phenomenal across the board from mailing lists, JIRA, code reviews
and helping us move to GitHub pull requests to contributing features, bug
fixes and code and documentation improvements.

Thank you for your contribution and welcome to Apache Kafka, Ismael!

-- 
Thanks,
Neha

Tell us how you're using Kafka and what you'd like to see in the future

[Neha Narkhede]

Folks,

We'd like to hear from community members about how you are using Kafka
today, and what sort of features you'd like to see in the future.

We put together this survey
 to gather your
feedback, which should take only a few minutes to complete.

Please be candid about what’s working and what needs to be improved—this
community is greater than the sum of its parts, and your responses will
ultimately help to steer the development of Kafka moving forward.

I will share the results of the survey for everyone to look at.

Thank you!

Best,
Neha

Re: 0.9.0.1 RC1

[Neha Narkhede]

+1 (binding).

Verified source and binary artifacts, ran ./gradlew testAll, quick start on
source artifact and Scala 2.11 binary artifact.

On Mon, Feb 15, 2016 at 7:43 PM, Ewen Cheslack-Postava 
wrote:

> Yeah, I saw
>
> kafka.network.SocketServerTest > tooBigRequestIsRejected FAILED
> java.net.SocketException: Broken pipe
> at java.net.SocketOutputStream.socketWrite0(Native Method)
> at
> java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:113)
> at java.net.SocketOutputStream.write(SocketOutputStream.java:138)
> at java.io.DataOutputStream.writeShort(DataOutputStream.java:168)
> at
> kafka.network.SocketServerTest.sendRequest(SocketServerTest.scala:62)
> at
>
> kafka.network.SocketServerTest.tooBigRequestIsRejected(SocketServerTest.scala:132)
>
> from the source artifacts one time.
>
> For reference, quick check of Confluent's kafka-trunk build
> http://jenkins.confluent.io/job/kafka-trunk/ doesn't show these specific
> transient errors (but it's also, obviously, a different branch). However,
> it does show other unrelated transient test errors.
>
> I've run the tests on trunk probably a dozen times in the past week for
> various PRs and not seen these test failures. The fact that I saw one the
> first time I ran tests on 0.9.0 has me a bit worried, though a couple of
> more test runs didn't have the same result.
>
> Also, for this specific test, I reopened
> https://issues.apache.org/jira/browse/KAFKA-2398 in August and then
> haven't
> seen it much since and we released the last version with that bug open...
>
> I guess I'm a wary +1 since we have a system test run passing, I've only
> seen this once, and it seems to be an existing transient test issue that
> had no impact in practice.
>
> -Ewen
>
>
> On Mon, Feb 15, 2016 at 8:39 PM, Ismael Juma  wrote:
>
> > +1 (non-binding).
> >
> > Verified source and binary artifacts, ran ./gradlew testAll with JDK
> 7u80,
> > quick start on source artifact and Scala 2.11 binary artifact.
> >
> > Ismael
> >
> > On Fri, Feb 12, 2016 at 2:55 AM, Jun Rao  wrote:
> >
> > > This is the first candidate for release of Apache Kafka 0.9.0.1. This a
> > bug
> > > fix release that fixes 70 issues.
> > >
> > > Release Notes for the 0.9.0.1 release
> > >
> >
> https://home.apache.org/~junrao/kafka-0.9.0.1-candidate1/RELEASE_NOTES.html
> > >
> > > *** Please download, test and vote by Tuesday, Feb. 16, 7pm PT
> > >
> > > Kafka's KEYS file containing PGP keys we use to sign the release:
> > > http://kafka.apache.org/KEYS in addition to the md5, sha1
> > > and sha2 (SHA256) checksum.
> > >
> > > * Release artifacts to be voted upon (source and binary):
> > > https://home.apache.org/~junrao/kafka-0.9.0.1-candidate1/
> > >
> > > * Maven artifacts to be voted upon prior to release:
> > > https://repository.apache.org/content/groups/staging/org/apache/kafka/
> > >
> > > * scala-doc
> > > https://home.apache.org/~junrao/kafka-0.9.0.1-candidate1/scaladoc/
> > >
> > > * java-doc
> > > https://home.apache.org/~junrao/kafka-0.9.0.1-candidate1/javadoc/
> > >
> > > * The tag to be voted upon (off the 0.9.0 branch) is the 0.9.0.1 tag
> > >
> > >
> >
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=2c17685a45efe665bf5f24c0296cb8f9e1157e89
> > >
> > > * Documentation
> > > http://kafka.apache.org/090/documentation.html
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> >
>
>
>
> --
> Thanks,
> Ewen
>



-- 
Thanks,
Neha

Re: 0.9.0.0 RC4

[Neha Narkhede]

+1 (binding).

Verified source and binary artifacts, ran unit tests.

On Mon, Nov 23, 2015 at 9:32 AM, Jun Rao  wrote:

> I updated the release notes. Since this doesn't affect the release
> artifacts to be voted upon, we don't have to do another RC.
>
> Please vote by 6pm PT today.
>
> Thanks,
>
> Jun
>
> On Mon, Nov 23, 2015 at 8:43 AM, Guozhang Wang  wrote:
>
> > I think we should update the release notes to remove the Kafka Streams
> > tickets, I have marked them as 0.9.1.0.
> >
> > Guozhang
> >
> > On Fri, Nov 20, 2015 at 5:21 PM, Jun Rao  wrote:
> >
> > > This is the fourth candidate for release of Apache Kafka 0.9.0.0. This
> a
> > > major release that includes (1) authentication (through SSL and SASL)
> and
> > > authorization, (2) a new java consumer, (3) a Kafka connect framework
> for
> > > data ingestion and egression, and (4) quotas. Since this is a major
> > > release, we will give people a bit more time for trying this out.
> > >
> > > Release Notes for the 0.9.0.0 release
> > >
> > >
> >
> https://people.apache.org/~junrao/kafka-0.9.0.0-candidate4/RELEASE_NOTES.html
> > >
> > > *** Please download, test and vote by Monday, Nov. 23, 6pm PT
> > >
> > > Kafka's KEYS file containing PGP keys we use to sign the release:
> > > http://kafka.apache.org/KEYS in addition to the md5, sha1
> > > and sha2 (SHA256) checksum.
> > >
> > > * Release artifacts to be voted upon (source and binary):
> > > https://people.apache.org/~junrao/kafka-0.9.0.0-candidate4/
> > >
> > > * Maven artifacts to be voted upon prior to release:
> > > https://repository.apache.org/content/groups/staging/
> > >
> > > * scala-doc
> > > https://people.apache.org/~junrao/kafka-0.9.0.0-candidate4/scaladoc/
> > >
> > > * java-doc
> > > https://people.apache.org/~junrao/kafka-0.9.0.0-candidate4/javadoc/
> > >
> > > * The tag to be voted upon (off the 0.9.0 branch) is the 0.9.0.0 tag
> > >
> > >
> >
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=132943b0f83831132cd46ac961cf6f1c00132565
> > >
> > > * Documentation
> > > http://kafka.apache.org/090/documentation.html
> > >
> > > /***
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
>



-- 
Thanks,
Neha

Re: [DISCUSSION] Kafka 0.8.2.2 release?

[Neha Narkhede]

How about looking at the scope for the 0.8.3 release first before we cut
yet another point release off of 0.8.2.2? Each release includes some
overhead and if there is a larger release in the works, it might be worth
working on getting that. My take is that the 2 things the community has
been waiting for is SSL support and the new consumer and we have been
promising to get 0.8.3 with both those features for several months now.

Looking at the progress on both, it seems we are very close to getting both
those checked in and it looks like we should get there in another 5-6
weeks. Furthermore, both of these features are large and I anticipate us
receiving feedback and bugs that will require a couple of point releases on
top of 0.8.3 anyway. One possibility is to work on 0.8.3 together now and
get the community to use the newly released features, gather feedback and
do point releases incorporating that feedback and iterate on it.

We could absolutely do both 0.8.2.2 and 0.8.3. What I'd ask for is for us
to look at the 0.8.3 timeline too and make a call whether 0.8.2.2 still
makes sense.

Thanks,
Neha

On Tue, Aug 18, 2015 at 10:24 AM, Gwen Shapira  wrote:

> Thanks Jun.
>
> I updated the list with your suggestions.
> If anyone feels we are missing a critical patch for 0.8.2.2, please speak
> up.
>
> Gwen
>
> On Mon, Aug 17, 2015 at 5:40 PM, Jun Rao  wrote:
>
> > Hi, Grant,
> >
> > I took a look at that list. None of those is really critical as you said.
> > So, I'd suggest that we not include those to minimize the scope of the
> > release.
> >
> > Thanks,
> >
> > Jun
> >
> > On Mon, Aug 17, 2015 at 5:16 PM, Grant Henke 
> wrote:
> >
> > > Thanks Gwen.
> > >
> > > I updated a few small things on the wiki page.
> > >
> > > Below is a list of jiras I think could also be marked as included. All
> of
> > > these, though not super critical, seem like fairly small and low risk
> > > changes that help avoid potentially confusing issues or errors for
> users.
> > >
> > > KAFKA-2012
> > > KAFKA-972
> > > KAFKA-2337 & KAFKA-2393
> > > KAFKA-1867
> > > KAFKA-2407
> > > KAFKA-2234
> > > KAFKA-1866
> > > KAFKA-2345 & KAFKA-2355
> > >
> > > thoughts?
> > >
> > > Thank you,
> > > Grant
> > >
> > > On Mon, Aug 17, 2015 at 4:56 PM, Gwen Shapira 
> wrote:
> > >
> > > > Thanks for creating a list, Grant!
> > > >
> > > > I placed it on the wiki with a quick evaluation of the content and
> > > whether
> > > > it should be in 0.8.2.2:
> > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/Proposed+patches+for+0.8.2.2
> > > >
> > > > I'm attempting to only cherrypick fixes that are both important for
> > large
> > > > number of users (or very critical to some users) and very safe
> (mostly
> > > > judged by the size of the change, but not only)
> > > >
> > > > If your favorite bugfix is missing from the list, or is there but
> > marked
> > > > "No", please let us know (in this thread) what we are missing and why
> > it
> > > is
> > > > both important and safe.
> > > > Also, if I accidentally included something you consider unsafe, speak
> > up!
> > > >
> > > > Gwen
> > > >
> > > > On Mon, Aug 17, 2015 at 8:17 AM, Grant Henke 
> > > wrote:
> > > >
> > > > > +dev
> > > > >
> > > > > Adding dev list back in. Somehow it got dropped.
> > > > >
> > > > >
> > > > > On Mon, Aug 17, 2015 at 10:16 AM, Grant Henke  >
> > > > wrote:
> > > > >
> > > > > > Below is a list of candidate bug fix jiras marked fixed for
> 0.8.3.
> > I
> > > > > don't
> > > > > > suspect all of these will (or should) make it into the release
> but
> > > this
> > > > > > should be a relatively complete list to work from:
> > > > > >
> > > > > >- KAFKA-2114 <
> https://issues.apache.org/jira/browse/KAFKA-2114
> > >:
> > > > > Unable
> > > > > >to change min.insync.replicas default
> > > > > >- KAFKA-1702 <
> https://issues.apache.org/jira/browse/KAFKA-1702
> > >:
> > > > > >Messages silently Lost by producer
> > > > > >- KAFKA-2012 <
> https://issues.apache.org/jira/browse/KAFKA-2012
> > >:
> > > > > >Broker should automatically handle corrupt index files
> > > > > >- KAFKA-2406 <
> https://issues.apache.org/jira/browse/KAFKA-2406
> > >:
> > > > ISR
> > > > > >propagation should be throttled to avoid overwhelming
> > controller.
> > > > > >- KAFKA-2336 <
> https://issues.apache.org/jira/browse/KAFKA-2336
> > >:
> > > > > >Changing offsets.topic.num.partitions after the offset topic
> is
> > > > > created
> > > > > >breaks consumer group partition assignment
> > > > > >- KAFKA-2337 <
> https://issues.apache.org/jira/browse/KAFKA-2337
> > >:
> > > > > Verify
> > > > > >that metric names will not collide when creating new topics
> > > > > >- KAFKA-2393 <
> https://issues.apache.org/jira/browse/KAFKA-2393
> > >:
> > > > > >Correctly Handle InvalidTopicException in
> > > > KafkaApis.getTopicMetadata()
> > > > > >- KAFKA-2189 <
> https://issues.apache.org/jira/browse/KAFKA-2189
> > >:
> > > > 

Help Us Nominate Apache Kafka for a 2015 Bossie (Best of OSS) Award - Due June 30th

[Neha Narkhede]

Hello Kafka community members,

We appreciate your use and support of Kafka and all the feedback you’ve
provided to us along the way.  If all is still going well with Kafka and
you’re realizing great value from it, we’d like your support in nominating
Kafka for a 2015 InfoWorld Bossie award, which is an annual award where
InfoWorld honors the best of open source software.


As a reminder, Kafka

was
selected as one of "InfoWorld's top picks in distributed data processing,
data analytics, machine learning, NoSQL databases, and the Hadoop
ecosystem." A technology can win consecutive years, so there's nothing
stopping Kafka from making the list again.

Nominations for this award are very simple and require you to simply
deliver an email to InfoWorld's executive editor Doug Dineley (
doug_dine...@infoworld.com) with the following information:

   -

   The name of your software, or your use case
   -

   A link to Kafka's website: http://kafka.apache.org/
   -

   A few sentences on how you or your customers are using the software and
   why it is important and award-worthy.

Submissions must be sent to Doug  by June 30,
2015. Please let us know if you have any questions or if we can help in any
way.


Thank you for being part of the Kafka community!


-- 
Best,
Neha

Re: [ANNOUNCEMENT] Apache Kafka 0.8.2.1 Released

[Neha Narkhede]

Thanks for driving this Jun and everyone for the contributions!

On Wed, Mar 11, 2015 at 12:01 PM, Jun Rao  wrote:

> The Apache Kafka community is pleased to announce the release for Apache
> Kafka 0.8.2.1.
>
> The 0.8.2.1 release fixes 4 critical issues in 0.8.2.0.
>
> All of the changes in this release can be found:
> https://archive.apache.org/dist/kafka/0.8.2.1/RELEASE_NOTES.html
>
> Apache Kafka is high-throughput, publish-subscribe messaging system
> rethought of as a distributed commit log.
>
> ** Fast => A single Kafka broker can handle hundreds of megabytes of reads
> and
> writes per second from thousands of clients.
>
> ** Scalable => Kafka is designed to allow a single cluster to serve as the
> central data backbone
> for a large organization. It can be elastically and transparently expanded
> without downtime.
> Data streams are partitioned and spread over a cluster of machines to
> allow data streams
> larger than the capability of any single machine and to allow clusters of
> co-ordinated consumers.
>
> ** Durable => Messages are persisted on disk and replicated within the
> cluster to prevent
> data loss. Each broker can handle terabytes of messages without
> performance impact.
>
> ** Distributed by Design => Kafka has a modern cluster-centric design that
> offers
> strong durability and fault-tolerance guarantees.
>
> You can download the release from: http://kafka.apache.org/downloads.html
>
> We welcome your help and feedback. For more information on how to
> report problems, and to get involved, visit the project website at
> http://kafka.apache.org/
>
> Thanks,
>
> Jun
>



-- 
Thanks,
Neha

Re: Trying to get kafka data to Hadoop

[Neha Narkhede]

Thanks Jagat for the callout!

Confluent Platform 1.0  includes Camus and we
were happy to address any questions in our community mailing list
.



On Wed, Mar 4, 2015 at 8:41 PM, max square  wrote:

> Thunder,
>
> thanks for your reply. The hadoop job is now correctly configured (the
> client was not getting the correct jars), however I am getting Avro
> formatting exceptions due to the format the schema-repo server follows. I
> think I will do something similar and create our own branch that uses the
> schema repo. Any gotchas you can advice on?
>
> Thanks!
>
> Max
>
> On Wed, Mar 4, 2015 at 9:24 PM, Thunder Stumpges 
> wrote:
>
> > What branch of camus are you using? We have our own fork that we updated
> > the camus dependency from the avro snapshot of the REST Schema Repository
> > to the new "official" one you mention in github.com/schema-repo. I was
> > not aware of a branch on the main linked-in camus repo that has this.
> >
> > That being said, we are doing essentially this same thing however we are
> > using a single shaded uber-jar. I believe the maven project builds this
> > automatically doesnt it?
> >
> > I'll take a look at the details of how we are invoking this on our site
> > and get back to you.
> >
> > Cheers,
> > Thunder
> >
> >
> > -Original Message-
> > From: max square [max2subscr...@gmail.com]
> > Received: Wednesday, 04 Mar 2015, 5:38PM
> > To: users@kafka.apache.org [users@kafka.apache.org]
> > Subject: Trying to get kafka data to Hadoop
> >
> > Hi all,
> >
> > I have browsed through different conversations around Camus, and bring
> this
> > as a kinda Kafka question. I know is not the most orthodox, but if
> someone
> > has some thoughts I'd appreciate ir.
> >
> > That said, I am trying to set up Camus, using a 3 node Kafka cluster
> > 0.8.2.1, using a project that is trying to build Avro Schema-Repo
> > . All of the Avro schemas
> for
> > the topics are published correctly. I am building Camus and using:
> >
> > hadoop jar camus-example-0.1.0-SNAPSHOT-shaded.jar
> com.linkedin.camus.etl.
> > kafka.CamusJob -libjars $CAMUS_LIBJARS  -D mapreduce.job.user.classpath.
> > first=true -P config.properties
> >
> > As the command to start the job, where I have set up an environment
> > variable that holds all the libjars that the mvn package command
> generates.
> >
> > I have also set the following properties to configure the job:
> > camus.message.decoder.class=com.linkedin.camus.etl.kafka.coders.
> > LatestSchemaKafkaAvroMessageDecoder
> >
> >
> kafka.message.coder.schema.registry.class=com.linkedin.camus.schemaregistry.
> > AvroRestSchemaRegistry
> >
> > etl.schema.registry.url=http://10.0.14.25:2876/schema-repo/
> >
> > When I execute the job I get an Exception indicating the
> > AvroRestSchemaRegistry class can't be found (I've double checked it's
> part
> > of the libjars). I wanted to ask if this is the correct way to set up
> this
> > integration, and if anyone has pointers on why the job is not finding the
> > class AvroRestSchemaRegistry
> >
> > Thanks in advance for the help!
> >
> > Max
> >
> > Follows the complete stack trace:
> >
> > [CamusJob] - failed to create decoder
> >
> > com.linkedin.camus.coders.MessageDecoderException:
> > com.linkedin.camus.coders
> > .MessageDecoderException:java.lang.ClassNotFoundException: com.linkedin.
> > camus.schemaregistry.AvroRestSchemaRegistry
> >at com.linkedin.camus.etl.kafka.coders.MessageDecoderFactory.
> > createMessageDecoder(MessageDecoderFactory.java:29)
> >
> >   at
> > com.linkedin.camus.etl.kafka.mapred.EtlInputFormat.createMessageDecoder
> > (EtlInputFormat.java:391)
> >
> >at com.linkedin.camus.etl.kafka.mapred.EtlInputFormat.getSplits(
> > EtlInputFormat.java:256)
> >
> >at
> org.apache.hadoop.mapred.JobClient.writeNewSplits(JobClient.java:
> > 1107)
> >
> >at
> > org.apache.hadoop.mapred.JobClient.writeSplits(JobClient.java:1124
> > )
> >
> >  at org.apache.hadoop.mapred.JobClient.access$600(JobClient.java:178)
> >
> >at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:1023)
> >
> >at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:976)
> >
> >at java.security.AccessController.doPrivileged(Native Method)
> >
> >at javax.security.auth.Subject.doAs(Subject.java:415)
> >
> >at org.apache.hadoop.security.UserGroupInformation.doAs(
> > UserGroupInformation.java:1642)
> >
> >at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.
> > java:976)
> >
> >at org.apache.hadoop.mapreduce.Job.submit(Job.java:582)
> >
> >at com.linkedin.camus.etl.kafka.CamusJob.run(CamusJob.java:335)
> >
> >at com.linkedin.camus.etl.kafka.CamusJob.run(CamusJob.java:563)
> >
> >at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)
> >
> >at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:84)
> >
> >at com.li

Re: [kafka-clients] Re: [VOTE] 0.8.2.1 Candidate 2

[Neha Narkhede]

+1. Verified quick start, unit tests.

On Tue, Mar 3, 2015 at 12:09 PM, Joe Stein  wrote:

> Ok, lets fix the transient test failure on trunk agreed not a blocker.
>
> +1 quick start passed, verified artifacts, updates in scala
> https://github.com/stealthly/scala-kafka/tree/0.8.2.1 and go
> https://github.com/stealthly/go_kafka_client/tree/0.8.2.1 look good
>
> ~ Joe Stein
> - - - - - - - - - - - - - - - - -
>
>   http://www.stealth.ly
> - - - - - - - - - - - - - - - - -
>
> On Tue, Mar 3, 2015 at 12:30 PM, Jun Rao  wrote:
>
> > Hi, Joe,
> >
> > Yes, that unit test does have transient failures from time to time. The
> > issue seems to be with the unit test itself and not the actual code. So,
> > this is not a blocker for 0.8.2.1 release. I think we can just fix it in
> > trunk.
> >
> > Thanks,
> >
> > Jun
> >
> > On Tue, Mar 3, 2015 at 9:08 AM, Joe Stein  wrote:
> >
> >> Jun, I have most everything looks good except I keep getting test
> >> failures from wget
> >>
> https://people.apache.org/~junrao/kafka-0.8.2.1-candidate2/kafka-0.8.2.1-src.tgz
> >> && tar -xvf kafka-0.8.2.1-src.tgz && cd kafka-0.8.2.1-src && gradle &&
> >> ./gradlew test
> >>
> >> kafka.api.ProducerFailureHandlingTest >
> >> testNotEnoughReplicasAfterBrokerShutdown FAILED
> >> org.scalatest.junit.JUnitTestFailedError: Expected
> >> NotEnoughReplicasException when producing to topic with fewer brokers
> than
> >> min.insync.replicas
> >> at
> >>
> org.scalatest.junit.AssertionsForJUnit$class.newAssertionFailedException(AssertionsForJUnit.scala:101)
> >> at
> >>
> org.scalatest.junit.JUnit3Suite.newAssertionFailedException(JUnit3Suite.scala:149)
> >> at org.scalatest.Assertions$class.fail(Assertions.scala:711)
> >> at org.scalatest.junit.JUnit3Suite.fail(JUnit3Suite.scala:149)
> >> at
> >>
> kafka.api.ProducerFailureHandlingTest.testNotEnoughReplicasAfterBrokerShutdown(ProducerFailureHandlingTest.scala:355)
> >>
> >> This happens to me all the time on a few different machines.
> >>
> >> ~ Joe Stein
> >> - - - - - - - - - - - - - - - - -
> >>
> >>   http://www.stealth.ly
> >> - - - - - - - - - - - - - - - - -
> >>
> >> On Mon, Mar 2, 2015 at 7:36 PM, Jun Rao  wrote:
> >>
> >>> +1 from me. Verified quickstart and unit tests.
> >>>
> >>> Thanks,
> >>>
> >>> Jun
> >>>
> >>> On Thu, Feb 26, 2015 at 2:59 PM, Jun Rao  wrote:
> >>>
>  This is the second candidate for release of Apache Kafka 0.8.2.1. This
>  fixes 4 critical issue in 0.8.2.0.
> 
>  Release Notes for the 0.8.2.1 release
> 
> 
> https://people.apache.org/~junrao/kafka-0.8.2.1-candidate2/RELEASE_NOTES.html
> 
>  *** Please download, test and vote by Monday, Mar 2, 3pm PT
> 
>  Kafka's KEYS file containing PGP keys we use to sign the release:
>  http://kafka.apache.org/KEYS in addition to the md5, sha1
>  and sha2 (SHA256) checksum.
> 
>  * Release artifacts to be voted upon (source and binary):
>  https://people.apache.org/~junrao/kafka-0.8.2.1-candidate2/
> 
>  * Maven artifacts to be voted upon prior to release:
>  https://repository.apache.org/content/groups/staging/
> 
>  * scala-doc
>  https://people.apache.org/~junrao/kafka-0.8.2.1-candidate2/scaladoc/
> 
>  * java-doc
>  https://people.apache.org/~junrao/kafka-0.8.2.1-candidate2/javadoc/
> 
>  * The tag to be voted upon (off the 0.8.2 branch) is the 0.8.2.1 tag
> 
> 
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=bd1bfb63ec73c10d08432ac893a23f28281ea021
>  (git commit ee1267b127f3081db491fa1bf9a287084c324e36)
> 
>  /***
> 
>  Thanks,
> 
>  Jun
> 
> 
> >>>  --
> >>> You received this message because you are subscribed to the Google
> >>> Groups "kafka-clients" group.
> >>> To unsubscribe from this group and stop receiving emails from it, send
> >>> an email to kafka-clients+unsubscr...@googlegroups.com.
> >>> To post to this group, send email to kafka-clie...@googlegroups.com.
> >>> Visit this group at http://groups.google.com/group/kafka-clients.
> >>> To view this discussion on the web visit
> >>>
> https://groups.google.com/d/msgid/kafka-clients/CAFc58G_5FuLKx3_kM4PCgqQL8d%2B4sqE0o-%2BXfu3FJicAgn5KPw%40mail.gmail.com
> >>> <
> https://groups.google.com/d/msgid/kafka-clients/CAFc58G_5FuLKx3_kM4PCgqQL8d%2B4sqE0o-%2BXfu3FJicAgn5KPw%40mail.gmail.com?utm_medium=email&utm_source=footer
> >
> >>> .
> >>>
> >>> For more options, visit https://groups.google.com/d/optout.
> >>>
> >>
> >>
> >
>



-- 
Thanks,
Neha

Re: Kafka Poll: Version You Use?

[Neha Narkhede]

Thanks for running the poll and sharing the results!

On Wed, Mar 4, 2015 at 8:34 PM, Otis Gospodnetic  wrote:

> Hi,
>
> You can see the number of voters in the poll itself (view poll results link
> in the poll widget).
> Audience details unknown, but the poll was posted on:
> * twitter - https://twitter.com/sematext/status/57050147435776
> * LinkedIn - a few groups - Kafka, DevOps, and I think another larger one
> * this mailing list
>
> Otis
> --
> Monitoring * Alerting * Anomaly Detection * Centralized Log Management
> Solr & Elasticsearch Support * http://sematext.com/
>
>
> On Wed, Mar 4, 2015 at 11:24 PM, Christian Csar  wrote:
>
> > Do you have a anything on the number of voters, or audience breakdown?
> >
> > Christian
> >
> > On Wed, Mar 4, 2015 at 8:08 PM, Otis Gospodnetic <
> > otis.gospodne...@gmail.com
> > > wrote:
> >
> > > Hello hello,
> > >
> > > Results of the poll are here!
> > > Any guesses before looking?
> > > What % of Kafka users are on 0.8.2.x already?
> > > What % of people are still on 0.7.x?
> > >
> > >
> > >
> >
> http://blog.sematext.com/2015/03/04/poll-results-kafka-version-distribution/
> > >
> > > Otis
> > > --
> > > Monitoring * Alerting * Anomaly Detection * Centralized Log Management
> > > Solr & Elasticsearch Support * http://sematext.com/
> > >
> > >
> > > On Thu, Feb 26, 2015 at 3:32 PM, Otis Gospodnetic <
> > > otis.gospodne...@gmail.com> wrote:
> > >
> > > > Hi,
> > > >
> > > > With 0.8.2 out I thought it might be useful for everyone to see which
> > > > version(s) of Kafka people are using.
> > > >
> > > > Here's a quick poll:
> > > > http://blog.sematext.com/2015/02/23/kafka-poll-version-you-use/
> > > >
> > > > We'll publish the results next week.
> > > >
> > > > Thanks,
> > > > Otis
> > > > --
> > > > Monitoring * Alerting * Anomaly Detection * Centralized Log
> Management
> > > > Solr & Elasticsearch Support * http://sematext.com/
> > > >
> > > >
> > >
> >
>



-- 
Thanks,
Neha

Announcing the Confluent Platform built on Apache Kafka

[Neha Narkhede]

Folks,

We, at Confluent , are excited to announce the release
of Confluent Platform 1.0 built around Apache Kafka -
http://blog.confluent.io/2015/02/25/announcing-the-confluent-platform-1-0/

We also published a detailed two-part guide on how you can put Kafka to use
in your organization -
http://blog.confluent.io/2015/02/25/stream-data-platform-1/

And, there is a public mailing list where we would love to hear your
feedback: confluent-platf...@googlegroups.com

Thanks,
Neha

Re: About Symantec's encryption-thru-Kafka proof of concept

[Neha Narkhede]

Great. Thanks for sharing!

On Thu, Feb 19, 2015 at 8:51 PM, Jim Hoagland 
wrote:

> Hi Folks,
>
> At the recent Kafka Meetup in Mountain View there was interest expressed
> about the encryption through Kafka proof of concept that Symantec did a
> few months ago, so I have created a blog post with some details about it.
> You can find that here:
>   http://goo.gl/sjYGWN
>
> Let me know if you have any thoughts or questions.
>
> Thanks,
>
>   Jim
>
> --
> Jim Hoagland, Ph.D.
> Sr. Principal Software Engineer
> Big Data Analytics Team
> Cloud Platform Engineering
> Symantec Corporation
> http://cpe.symantec.com
>
>


-- 
Thanks,
Neha

Re: Kafka Leader Rebalance

[Neha Narkhede]

Are you are on 0.8.1.1 or higher and are you using the kafka-topics tool to
see which leaders are on which brokers?
Did you also wait for more than 15 seconds to see if the leader election
takes place?
Are there any errors in the controller log?

On Sat, Feb 14, 2015 at 9:35 AM, nitin sharma 
wrote:

> Hi,
>
> Anyone who knows how to solve my problem, please respond..
>
> Regards,
> Nitin Kumar Sharma.
>
>
> On Thu, Feb 12, 2015 at 3:50 PM, nitin sharma  >
> wrote:
>
> > Hi Team,
> >
> > i need some help in solving my current issue related to
> > "kafka-leadership-rebalance"
> >
> > I have 2 brokers..  i have deployed 2 topics with 2 partition and 2
> > replica each in following format.. I made use of kafka-reassignment.sh
> for
> > same
> >
> > Topic   partition   Leader Follower
> > Topic#1 P#0 Broker#1Broker#2
> > Topic#1 P#1 Broker#2Broker#1
> >
> > Topic#2 P#0Broker#1 Broker#2
> > Topic#2 P#1 Broker#2Broker#1
> >
> >
> > By problem is when a particular Broker server restarts, the complete
> > arrangement goes for toss. All the partitions moves to one server.
> >
> > I have tried putting following properties but nothing has worked:
> > auto.leader.rebalance.enable=true
> > leader.imbalance.check.interval.seconds=15
> > leader.imbalance.per.broker.percenatage=1
> >
> >
> > Regards,
> > Nitin Kumar Sharma.
> >
> >
>



-- 
Thanks,
Neha

Re: [kafka-clients] Re: [VOTE] 0.8.2.0 Candidate 3

[Neha Narkhede]

Great! Thanks Jun for helping with the release and everyone involved for
your contributions.

On Mon, Feb 2, 2015 at 1:32 PM, Joe Stein  wrote:

> Huzzah!
>
> Thanks Jun for preparing the release candidates and getting this out to the
> community.
>
> - Joe Stein
>
> On Mon, Feb 2, 2015 at 2:27 PM, Jun Rao  wrote:
>
> > The following are the results of the votes.
> >
> > +1 binding = 3 votes
> > +1 non-binding = 1 votes
> > -1 = 0 votes
> > 0 = 0 votes
> >
> > The vote passes.
> >
> > I will release artifacts to maven central, update the dist svn and
> download
> > site. Will send out an announce after that.
> >
> > Thanks everyone that contributed to the work in 0.8.2.0!
> >
> > Jun
> >
> > On Wed, Jan 28, 2015 at 9:22 PM, Jun Rao  wrote:
> >
> >> This is the third candidate for release of Apache Kafka 0.8.2.0.
> >>
> >> Release Notes for the 0.8.2.0 release
> >>
> >>
> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate3/RELEASE_NOTES.html
> >>
> >> *** Please download, test and vote by Saturday, Jan 31, 11:30pm PT
> >>
> >> Kafka's KEYS file containing PGP keys we use to sign the release:
> >> http://kafka.apache.org/KEYS in addition to the md5, sha1 and sha2
> >> (SHA256) checksum.
> >>
> >> * Release artifacts to be voted upon (source and binary):
> >> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate3/
> >>
> >> * Maven artifacts to be voted upon prior to release:
> >> https://repository.apache.org/content/groups/staging/
> >>
> >> * scala-doc
> >> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate3/scaladoc/
> >>
> >> * java-doc
> >> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate3/javadoc/
> >>
> >> * The tag to be voted upon (off the 0.8.2 branch) is the 0.8.2.0 tag
> >>
> >>
> https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=223ac42a7a2a0dab378cc411f4938a9cea1eb7ea
> >> (commit 7130da90a9ee9e6fb4beb2a2a6ab05c06c9bfac4)
> >>
> >> /***
> >>
> >> Thanks,
> >>
> >> Jun
> >>
> >>
> >  --
> > You received this message because you are subscribed to the Google Groups
> > "kafka-clients" group.
> > To unsubscribe from this group and stop receiving emails from it, send an
> > email to kafka-clients+unsubscr...@googlegroups.com.
> > To post to this group, send email to kafka-clie...@googlegroups.com.
> > Visit this group at http://groups.google.com/group/kafka-clients.
> > To view this discussion on the web visit
> >
> https://groups.google.com/d/msgid/kafka-clients/CAFc58G-XRpw9ik35%2BCmsYm-uc2hjHet6fOpw4bF90ka9Z%3DhH%3Dw%40mail.gmail.com
> > <
> https://groups.google.com/d/msgid/kafka-clients/CAFc58G-XRpw9ik35%2BCmsYm-uc2hjHet6fOpw4bF90ka9Z%3DhH%3Dw%40mail.gmail.com?utm_medium=email&utm_source=footer
> >
> > .
> >
> > For more options, visit https://groups.google.com/d/optout.
> >
>



-- 
Thanks,
Neha

Re: [VOTE] 0.8.2.0 Candidate 3

[Neha Narkhede]

+1 (binding). Verified quickstart and unit tests ran ok.

On Sun, Feb 1, 2015 at 9:11 AM, Steven Wu  wrote:

> In Netflix, we have been using route53 DNS name as bootstrap servers in AWS
> env. Basically, when a kafka broker start, we add it to route53 DNS name
> for the cluster. this is like the VIP that Jay suggested.
>
> But we are also moving toward to use Eureka service registry for
> bootstrapping. We are worried that if DNS name happens to resolve to a bad
> broker. it might impact the bootstrap process/resiliency. We want to get a
> list of brokers from Eureka to pass in as "bootstrap.servers".
>
>
>
> On Sun, Feb 1, 2015 at 5:30 AM, Jay Kreps  wrote:
>
> > You may already know this but the producer doesn't require a complete
> list
> > of brokers in its config, it just requires the connection info for one
> > active broker which it uses to discover the rest of the brokers. We allow
> > you to specify multiple urls here for failover in cases where you aren't
> > using a vip. So if you can put three brokers into the VIP for metadata
> > bootstrapping you can still scale up and down the rest of the cluster.
> >
> > -Jay
> >
> > On Sun, Feb 1, 2015 at 12:17 AM, Alex The Rocker 
> > wrote:
> >
> > > Jun:
> > >
> > > You raise a very good question: let me explain why we use
> > > Broker.getConnectionString(), so may be we'll get a supported way to
> > > answer our need.
> > >
> > > We use Broker.getConnectionString() because we deploy "Kafka services"
> > > in Amazon EC2 with the following architecture:
> > > * Three VMs dedicated to Zookeeper processes
> > > * At least two VMs with Kafka broker, but depending on load it can be
> > > scaled to more broker VMs. Brokers self-register their address in
> > > Zookeeper by serializing Broker objects in Zk.
> > >
> > > The VMs with Zookeeper have Elastic IPs = stable public IPs,
> > >
> > > These public IPs are fed to the  various "Application services" which
> > > rely on Kafka to stream their logs & monitoring data to our central
> > > Hadoop system.
> > >
> > > Using zkclient and the above mentionned public zookeeper IPs, we get
> > > the list of brokers registrered to a given "Kafka service":  this is
> > > where we unserializer Broker objects and then use
> > > getConnectionString() to discover the brokers' addresses. Then,
> > > brokers addresses are used to initialize the Kafka producer(s).
> > >
> > > The whole trick is that we cannot use Elastic IP (=stable IPs) for
> > > Kafka VMs, because of their 'elastic" nature : we want to be able to
> > > scale up / down the number of VMs with Kafka brokers.
> > >
> > > Now, we understand that using non public Kafka API is bad : we've been
> > > broken when moving to 0.8.1.1, then again when moving to 0.8.2.0...
> > >
> > > So it's time to raise the right question: what would be the supported
> > > way to configure our producers given our dynamic-IP-for-brokers
> > > context?
> > >
> > > Thanks,
> > > Alex.
> > >
> > > 2015-02-01 8:55 GMT+01:00 VERMEERBERGEN Alexandre
> > > :
> > > >
> > > > -Original Message-
> > > > From: Jun Rao [mailto:j...@confluent.io]
> > > > Sent: Sunday, February 01, 2015 3:03
> > > > To: users@kafka.apache.org; kafka-clie...@googlegroups.com
> > > > Cc: d...@kafka.apache.org
> > > > Subject: Re: [VOTE] 0.8.2.0 Candidate 3
> > > >
> > > > Hi, Alex,
> > > >
> > > > Thanks for testing RC3.
> > > >
> > > > Broker.connectionString() is actually not part of the public api for
> > the
> > > producer. Is there a particular reason that you need to use this api?
> > > >
> > > > Thanks,
> > > >
> > > > Jun
> > > >
> > > > On Sat, Jan 31, 2015 at 1:53 PM, Alex The Rocker <
> alex.m3...@gmail.com
> > >
> > > > wrote:
> > > >
> > > >> Hello,
> > > >>
> > > >> I have read Broker.scala source code, and I found the answer:
> > > >>  - With Kafka 0.8.1.1 we used Broker.getConnectionString() in our
> Java
> > > >> code.
> > > >>  - With Kafka 0.8.2.0, this method has been replaced by a 0-arity
> > > >> method without the "get" prefix, so we have to change our Java code
> to
> > > >> call
> > > >> Broker.connectionString()
> > > >>
> > > >> So despite binary compatibility is broken, we have a by-pass.
> > > >> I hope this will help other people relying on this API...
> > > >>
> > > >> and I'm going to continue tests with 0.8.2 rc3..
> > > >>
> > > >> Alex
> > > >>
> > > >> 2015-01-31 21:23 GMT+01:00 Alex The Rocker :
> > > >>
> > > >> > Hello,
> > > >> >
> > > >> > I ran my own tests made with kafka_2.10-0.8.1.1.tgz binaries with
> > > >> > our
> > > >> > application:
> > > >> >
> > > >> > 1st test:
> > > >> > ==
> > > >> >   replace all kafka .jar files in our application on consumming
> side
> > > >> >   (without recompiling anything)
> > > >> >   => tests passed, OK
> > > >> >
> > > >> > 2nd test:
> > > >> > ===
> > > >> >   replace all kafka .jar files in our application on producubg
> side
> > > >> >   (without recompiling anything)
> > > >> >   => KO, we get this error:
> > > >> >
>

Re: [kafka-clients] Re: [VOTE] 0.8.2.0 Candidate 2 (with the correct links)

[Neha Narkhede]

+1 (binding)
Verified keys, quick start, unit tests.

On Sat, Jan 24, 2015 at 4:26 PM, Joe Stein  wrote:

> That makes sense, thanks!
>
> On Sat, Jan 24, 2015 at 7:00 PM, Jay Kreps  wrote:
>
> > But I think the flaw in trying to guess what kind of serializer they will
> > use is when we get it wrong. Basically let's say we guess "String". Say
> 30%
> > of the time we will be right and we will save the two configuration
> lines.
> > 70% of the time we will be wrong and the user gets a super cryptic
> > ClassCastException: "xyz cannot be cast to [B" (because [B is how java
> > chooses to display the byte array class just to up the pain), then they
> > figure out how to subscribe to our mailing list and email us the cryptic
> > exception, then we explain about how we helpfully set these properties
> for
> > them to save them time. :-)
> >
> > https://www.google.com/?gws_rd=ssl#q=kafka+classcastexception+%22%5BB%22
> >
> > I think basically we did this experiment with the old clients and the
> > conclusion is that serialization is something you basically have to think
> > about to use Kafka and trying to guess just makes things worse.
> >
> > -Jay
> >
> > On Sat, Jan 24, 2015 at 2:51 PM, Joe Stein  wrote:
> >
> >> Maybe. I think the StringSerialzer could look more like a typical type
> of
> >> message.  Instead of encoding being a property it would be more
> typically
> >> just written in the bytes.
> >>
> >> On Sat, Jan 24, 2015 at 12:12 AM, Jay Kreps 
> wrote:
> >>
> >> > I don't think so--see if you buy my explanation. We previously
> defaulted
> >> > to the byte array serializer and it was a source of unending
> frustration
> >> > and confusion. Since it wasn't a required config people just went
> along
> >> > plugging in whatever objects they had, and thinking that changing the
> >> > parametric types would somehow help. Then they would get a class case
> >> > exception and assume our stuff was somehow busted, not realizing we
> had
> >> > helpfully configured a type different from what they were passing in
> >> under
> >> > the covers. So I think it is actually good for people to think: how
> am I
> >> > serializing my data, and getting that exception will make them ask
> that
> >> > question right?
> >> >
> >> > -Jay
> >> >
> >> > On Fri, Jan 23, 2015 at 9:06 PM, Joe Stein 
> >> wrote:
> >> >
> >> >> Should value.serializer in the new java producer be defaulted to
> >> >> Array[Byte] ?
> >> >>
> >> >> I was working on testing some upgrade paths and got this
> >> >>
> >> >> ! return exception in callback when buffer cannot accept message
> >> >>
> >> >>   ConfigException: Missing required configuration
> >> "value.serializer"
> >> >> which has no default value. (ConfigDef.java:124)
> >> >>
> >> >>   org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:48)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:235)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:129)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> ly.stealth.testing.BaseSpec$class.createNewKafkaProducer(BaseSpec.scala:42)
> >> >>
> >> >>
> >>  ly.stealth.testing.KafkaSpec.createNewKafkaProducer(KafkaSpec.scala:36)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> ly.stealth.testing.KafkaSpec$$anonfun$3$$anonfun$apply$37.apply(KafkaSpec.scala:175)
> >> >>
> >> >>
> >> >>
> >> >>
> >>
> ly.stealth.testing.KafkaSpec$$anonfun$3$$anonfun$apply$37.apply(KafkaSpec.scala:170)
> >> >>
> >> >>
> >> >>
> >> >> On Fri, Jan 23, 2015 at 5:55 PM, Jun Rao  wrote:
> >> >>
> >> >> > This is a reminder that the deadline for the vote is this Monday,
> Jan
> >> >> 26,
> >> >> > 7pm PT.
> >> >> >
> >> >> > Thanks,
> >> >> >
> >> >> > Jun
> >> >> >
> >> >> > On Wed, Jan 21, 2015 at 8:28 AM, Jun Rao  wrote:
> >> >> >
> >> >> >> This is the second candidate for release of Apache Kafka 0.8.2.0.
> >> There
> >> >> >> has been some changes since the 0.8.2 beta release, especially in
> >> the
> >> >> new
> >> >> >> java producer api and jmx mbean names. It would be great if people
> >> can
> >> >> test
> >> >> >> this out thoroughly.
> >> >> >>
> >> >> >> Release Notes for the 0.8.2.0 release
> >> >> >>
> >> >> >>
> >> >>
> >>
> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate2/RELEASE_NOTES.html
> >> >> >>
> >> >> >> *** Please download, test and vote by Monday, Jan 26h, 7pm PT
> >> >> >>
> >> >> >> Kafka's KEYS file containing PGP keys we use to sign the release:
> >> >> >> http://kafka.apache.org/KEYS in addition to the md5, sha1 and
> sha2
> >> >> >> (SHA256) checksum.
> >> >> >>
> >> >> >> * Release artifacts to be voted upon (source and binary):
> >> >> >> https://people.apache.org/~junrao/kafka-0.8.2.0-candidate2/
> >> >> >>
> >> >> >> * Maven artifacts to be voted upon prior to release:
> >> >> >> https://repository.apache.org/conten

Re: Release Date for Kafka 0.8.2

[Neha Narkhede]

You can track the blockers here
.
We are waiting on follow up patches for 2 JIRAs which are under review. We
should be able to release 0.8.2 in a few days.



On Mon, Jan 5, 2015 at 10:26 PM, Otis Gospodnetic <
otis.gospodne...@gmail.com> wrote:

> Hi Srividhya,
>
> See
>
> http://search-hadoop.com/m/4TaT4B9tys1/&subj=Re+Kafka+0+8+2+release+before+Santa+Claus
>
> Otis
> --
> Monitoring * Alerting * Anomaly Detection * Centralized Log Management
> Solr & Elasticsearch Support * http://sematext.com/
>
>
> On Mon, Jan 5, 2015 at 11:55 AM, Srividhya Shanmugam <
> srividhyashanmu...@fico.com> wrote:
>
> > Kafka Team,
> >
> > We are currently using the 0.8.2 beta version with a patch for
> KAFKA-1738.
> > Do you have any updates on when 0.8.2 final version will be released?
> >
> > Thanks,
> > Srividhya
> >
> >
> > This email and any files transmitted with it are confidential,
> proprietary
> > and intended solely for the individual or entity to whom they are
> > addressed. If you have received this email in error please delete it
> > immediately.
> >
>



-- 
Thanks,
Neha

Re: leader and isr were not set when create the topic

[Neha Narkhede]

It seems that somehow the follower lost its highwatermark checkpoint file.
Can you share the steps to reproduce this along with the Kafka versions you
are using?

On Mon, Dec 22, 2014 at 4:17 PM, Sa Li  wrote:

> I have three nodes: 100, 101, and 102
>
> When I restart all of them, seems now everything is ok, but I would like to
> paste the error messages I got from server.log from each node, see if you
> can help to understand what is the problem.
>
> on node 100
> [2014-12-23 00:04:39,401] ERROR [KafkaApi-100] Error when processing fetch
> request for partition [perf_producer_p8_test,7] offset 125000 from follower
> with correlation id 0 (kafka.server.KafkaApis)
> kafka.common.OffsetOutOfRangeException: Request for offset 125000 but we
> only have log segments in the range 0 to 0.
>  at kafka.log.Log.read(Log.scala:380)
>  at
>
> kafka.server.KafkaApis.kafka$server$KafkaApis$$readMessageSet(KafkaApis.scala:530)
>  at
>
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$readMessageSets$1.apply(KafkaApis.scala:476)
>
>  at
>
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$readMessageSets$1.apply(KafkaApis.scala:471)
>
>  at
>
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>  at
>
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>
>  at
> scala.collection.immutable.Map$Map3.foreach(Map.scala:154)
>  at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
>  at
> scala.collection.AbstractTraversable.map(Traversable.scala:105)
> ..
> ..
>
>
> in Node 101 and 102
> [2014-12-23 00:04:39,440] ERROR [ReplicaFetcherThread-0-100], Current
> offset 1 25000 for partition [perf_producer_p8_test,1] out of range; reset
> offset to 0 (kafka.server.ReplicaFetcherThread)
> [2014-12-23 00:04:39,442] INFO Truncating log perf_producer_p8_test-7 to
> offset 0. (kafka.log.Log)
> [2014-12-23 00:04:39,452] WARN [ReplicaFetcherThread-0-100], Replica 102
> for partition [perf_producer_p8_test,7] reset its fetch offset to current
> leader 100's latest offset 0 (kafka.server.ReplicaFetcherThread)
>
>
>
>
>
>
> On Mon, Dec 22, 2014 at 3:55 PM, Sa Li  wrote:
> >
> > Hello, Neha
> >
> > This is the error from server.log
> >
> > [2014-12-22 23:53:25,663] WARN [KafkaApi-100] Fetch request with
> > correlation id 1227732 from client ReplicaFetcherThread-0-100 on
> partition
> > [perf_producer_p8_test,1] failed due to Leader not local for partition
> > [perf_producer_p8_test,1] on broker 100 (kafka.server.KafkaApis)
> >
> >
> > On Mon, Dec 22, 2014 at 3:50 PM, Sa Li  wrote:
> >>
> >> I restart the kafka server, it is the same thing, sometime nothing
> listed
> >> on ISR, leader, I checked the state-change log
> >>
> >> [2014-12-22 23:46:38,164] TRACE Broker 100 cached leader info
> >>
> (LeaderAndIsrInfo:(Leader:101,ISR:101,102,100,LeaderEpoch:0,ControllerEpoch:4),ReplicationFactor:3),AllReplicas:101,102,100)
> >> for partition [perf_producer_p8_test,1] in response to UpdateMetadata
> >> request sent by controller 101 epoch 4 with correlation id 138
> >> (state.change.logger)
> >>
> >>
> >>
> >> On Mon, Dec 22, 2014 at 2:46 PM, Sa Li  wrote:
> >>
> >>> Hi, All
> >>>
> >>> I created a topic with 3 replications and 6 partitions, but when I
> check
> >>> this topic, seems there is no leader and isr were set for this topic,
> see
> >>>
> >>> bin/kafka-topics.sh --create --zookeeper 10.100.98.100:2181
> >>> --replication-factor 3 --partitions 6 --topic perf_producer_p6_test
> >>> SLF4J: Class path contains multiple SLF4J bindings.
> >>> SLF4J: Found binding in
> >>>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >>> SLF4J: Found binding in
> >>>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >>> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> >>> explanation.
> >>> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> >>> Created topic "perf_producer_p6_test".
> >>>
> >>> root@precise64:/etc/kafka# bin/kafka-topics.sh --describe --zookeeper
> >>> 10.100.98.100:2181 --topic perf_producer_p6_test
> >>> SLF4J: Class path contains multiple SLF4J bindings.
> >>> SLF4J: Found binding in
> >>>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >>> SLF4J: Found binding in
> >>>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >>> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> >>> explanation.
> >>> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> >>> Topic:pe

Re: Kafka 0.8.2 release - before Santa Claus?

[Neha Narkhede]

Went through the list and cleaned it up. Most patches just need a final
review. The only JIRA that has no patch is
https://issues.apache.org/jira/browse/KAFKA-1723.

On Fri, Dec 26, 2014 at 2:18 PM, Gwen Shapira  wrote:

> Actually, KAFKA-1785 <https://issues.apache.org/jira/browse/KAFKA-1785>
> can
> also wait - since it is likely to be part of a larger patch.
>
> On Thu, Dec 25, 2014 at 10:39 AM, Gwen Shapira 
> wrote:
>
> > IMO:
> > KAFKA-1790 - can be pushed out (or even marked as "won't fix")
> > KAFKA-1782 - can be pushed out (not really a blocker)
> >
> > The rest look like actual blockers to me.
> >
> > Gwen
> >
> > On Tue, Dec 23, 2014 at 1:32 PM, Otis Gospodnetic <
> > otis.gospodne...@gmail.com> wrote:
> >
> >> Hi,
> >>
> >> I see 16 open issues for 0.8.2 at
> >>
> >>
> https://issues.apache.org/jira/issues/?jql=project%20%3D%20KAFKA%20AND%20fixVersion%20%3D%200.8.2%20AND%20resolution%20%3D%20Unresolved%20ORDER%20BY%20updated%20DESC%2C%20due%20ASC%2C%20priority%20DESC%2C%20created%20ASC
> >> - some with patches, some blockers, some blockers without patches.
> >>
> >> Are all issues listed as blockers truly blockers for 0.8.2?
> >>
> >> Thanks,
> >> Otis
> >> --
> >> Monitoring * Alerting * Anomaly Detection * Centralized Log Management
> >> Solr & Elasticsearch Support * http://sematext.com/
> >>
> >>
> >> On Mon, Dec 1, 2014 at 8:13 PM, Joe Stein  wrote:
> >>
> >> > If we can have a build ready by Dec 26th I think that is feasible. I
> >> could
> >> > prepare and post that if we think we have the votes and a stable
> >> version.
> >> >
> >> > /***
> >> > Joe Stein
> >> > Founder, Principal Consultant
> >> > Big Data Open Source Security LLC
> >> > http://www.stealth.ly
> >> > Twitter: @allthingshadoop
> >> > /
> >> > On Dec 1, 2014 7:56 PM, "Neha Narkhede" 
> >> wrote:
> >> >
> >> > > +1 for doing a 0.8.2 final before the December break.
> >> > >
> >> > > On Mon, Dec 1, 2014 at 8:40 AM, Jun Rao  wrote:
> >> > >
> >> > > > We are currently discussing a last-minute API change to the new
> java
> >> > > > producer. We have also accumulated a few more 0.8.2 blockers.
> >> > > >
> >> > > >
> >> > > >
> >> > >
> >> >
> >>
> https://issues.apache.org/jira/browse/KAFKA-1642?jql=project%20%3D%20KAFKA%20AND%20status%20in%20(Open%2C%20%22In%20Progress%22%2C%20Reopened%2C%20%22Patch%20Available%22)%20AND%20priority%20%3D%20Blocker%20AND%20fixVersion%20%3D%200.8.2
> >> > > >
> >> > > > So, we likely will have another 0.8.2 release in Dec. However, I
> am
> >> not
> >> > > > sure if that's beta2 or final.
> >> > > >
> >> > > > Thanks,
> >> > > >
> >> > > > Jun
> >> > > >
> >> > > > On Wed, Nov 26, 2014 at 12:22 PM, Otis Gospodnetic <
> >> > > > otis.gospodne...@gmail.com> wrote:
> >> > > >
> >> > > > > Hi,
> >> > > > >
> >> > > > > People using SPM to monitor Kafka have been anxiously asking us
> >> about
> >> > > the
> >> > > > > 0.8.2 release and we've been telling them December.  Is that
> still
> >> > the
> >> > > > > plan?
> >> > > > >
> >> > > > > Thanks,
> >> > > > > Otis
> >> > > > > --
> >> > > > > Monitoring * Alerting * Anomaly Detection * Centralized Log
> >> > Management
> >> > > > > Solr & Elasticsearch Support * http://sematext.com/
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
> >
> >
>



-- 
Thanks,
Neha

Re: NPE in debug logging statement in kafka new producer

[Neha Narkhede]

>
> So, do I need to expect 0.8.2 official release will have generic
> Producer interface?


Yes

On Sun, Dec 28, 2014 at 11:31 PM, Bae, Jae Hyeon  wrote:

> I forgot to mention I was building with kafka-clients-0.8.2-beta.
>
> To reproduce this problem, I cloned kafka github repo and changed the
> source tree to the tag:0.8.2-beta and modified the dependency in my IDE
> from kafka-clients-0.8.2-beta to the source tree with tag:0.8.2-beta but I
> couldn't reproduce it... weird.
>
> Also, I found out in 0.8.2 branch Producer interface was changed with
> generics. So, do I need to expect 0.8.2 official release will have generic
> Producer interface?
>
> On Sun, Dec 28, 2014 at 11:57 AM, Neha Narkhede 
> wrote:
>
> > This is a bug. Would you mind filing a JIRA? Also feel free to upload a
> > patch.
> >
> > On Sat, Dec 27, 2014 at 7:25 PM, Bae, Jae Hyeon 
> > wrote:
> >
> > > Hi
> > >
> > > While I am testing kafka java producer, I saw the following NPE
> > >
> > > SLF4J: Failed toString() invocation on an object of type
> > > [org.apache.kafka.common.Cluster]
> > > java.lang.NullPointerException
> > > at
> org.apache.kafka.common.PartitionInfo.toString(PartitionInfo.java:72)
> > > at java.lang.String.valueOf(String.java:2854)
> > > at java.lang.StringBuilder.append(StringBuilder.java:128)
> > > at java.util.AbstractCollection.toString(AbstractCollection.java:458)
> > > at java.lang.String.valueOf(String.java:2854)
> > > at java.lang.StringBuilder.append(StringBuilder.java:128)
> > > at org.apache.kafka.common.Cluster.toString(Cluster.java:151)
> > > at
> > >
> > >
> >
> org.slf4j.helpers.MessageFormatter.safeObjectAppend(MessageFormatter.java:305)
> > > at
> > >
> > >
> >
> org.slf4j.helpers.MessageFormatter.deeplyAppendParameter(MessageFormatter.java:277)
> > > at
> > >
> org.slf4j.helpers.MessageFormatter.arrayFormat(MessageFormatter.java:231)
> > > at org.slf4j.helpers.MessageFormatter.format(MessageFormatter.java:152)
> > > at org.slf4j.impl.Log4jLoggerAdapter.debug(Log4jLoggerAdapter.java:251)
> > > at
> > >
> > >
> >
> org.apache.kafka.clients.producer.internals.Metadata.update(Metadata.java:133)
> > > at
> > >
> > >
> >
> org.apache.kafka.clients.NetworkClient.handleMetadataResponse(NetworkClient.java:299)
> > > at
> > >
> > >
> >
> org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:284)
> > > at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:185)
> > > at
> > org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:175)
> > > at
> > org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:115)
> > > at java.lang.Thread.run(Thread.java:744)
> > >
> > > This looks like not critical because it's debug() statement but this
> NPE
> > > means potential threat in kafka producer stableness.
> > >
> > > Do you have any idea? If you want, I can share my test code.
> > >
> > > Thank you
> > > Best, Jae
> > >
> >
>



-- 
Thanks,
Neha

Re: NPE in debug logging statement in kafka new producer

[Neha Narkhede]

This is a bug. Would you mind filing a JIRA? Also feel free to upload a
patch.

On Sat, Dec 27, 2014 at 7:25 PM, Bae, Jae Hyeon  wrote:

> Hi
>
> While I am testing kafka java producer, I saw the following NPE
>
> SLF4J: Failed toString() invocation on an object of type
> [org.apache.kafka.common.Cluster]
> java.lang.NullPointerException
> at org.apache.kafka.common.PartitionInfo.toString(PartitionInfo.java:72)
> at java.lang.String.valueOf(String.java:2854)
> at java.lang.StringBuilder.append(StringBuilder.java:128)
> at java.util.AbstractCollection.toString(AbstractCollection.java:458)
> at java.lang.String.valueOf(String.java:2854)
> at java.lang.StringBuilder.append(StringBuilder.java:128)
> at org.apache.kafka.common.Cluster.toString(Cluster.java:151)
> at
>
> org.slf4j.helpers.MessageFormatter.safeObjectAppend(MessageFormatter.java:305)
> at
>
> org.slf4j.helpers.MessageFormatter.deeplyAppendParameter(MessageFormatter.java:277)
> at
> org.slf4j.helpers.MessageFormatter.arrayFormat(MessageFormatter.java:231)
> at org.slf4j.helpers.MessageFormatter.format(MessageFormatter.java:152)
> at org.slf4j.impl.Log4jLoggerAdapter.debug(Log4jLoggerAdapter.java:251)
> at
>
> org.apache.kafka.clients.producer.internals.Metadata.update(Metadata.java:133)
> at
>
> org.apache.kafka.clients.NetworkClient.handleMetadataResponse(NetworkClient.java:299)
> at
>
> org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:284)
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:185)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:175)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:115)
> at java.lang.Thread.run(Thread.java:744)
>
> This looks like not critical because it's debug() statement but this NPE
> means potential threat in kafka producer stableness.
>
> Do you have any idea? If you want, I can share my test code.
>
> Thank you
> Best, Jae
>

Re: Rebalance not happening even after increasing max retries causing conflict in ZK

[Neha Narkhede]

Can you share a reproducible test case?

On Tue, Dec 9, 2014 at 7:11 AM, Mohit Kathuria 
wrote:

> Neha,
>
> The same issue reoccured with just 2 consumer processes. The exception was
> related to conflict in writing the ephemeral node. Below was the exception.
> Topic name is
>  "lst_plugin_com.spr.listening.plugin.impl.plugins.SemantriaEnrichmentPlugin"
> with 30 partitions. The 2 processes were running on 2 servers with ips
> 10.0.8.222 and 10.0.8.225.
>
> *2014-12-09 13:22:11 k.u.ZkUtils$ [INFO] I wrote this conflicted ephemeral
> node
> [{"version":1,"subscription":{"lst_plugin_com.spr.listening.plugin.impl.plugins.SemantriaEnrichmentPlugin":5},"pattern":"static","timestamp":"1417964160024"}]
> at
> /consumers/lst_plugin_com.spr.listening.plugin.impl.plugins.SemantriaEnrichmentPlugin/ids/lst_plugin_com.spr.listening.plugin.impl.plugins.SemantriaEnrichmentPlugin_ip-10-0-8-222-1417963753598-b19de58d
> a while back in a different session, hence I will backoff for this node to
> be deleted by Zookeeper and retry*
> Attached the complete error logs. The exception occured after the
> rebalance failed even after 40 retries. Rebalance failed as the process
> already owning some of the partitions did not give us ownership due to
> conflicting ephemeral nodes. As you suggested, we ran the wchp command  on
> the 3 zookeeper nodes at this time and figured out that the watcher was
> registered for only one of the process. I am copying the kafka consumer
> watcher registered on one of the zookeeper servers. (Attached are the wchp
> outputs of all 3 zk servers)
>
> *$echo wchp | nc localhost 2181 *
>
>
> */kafka/consumers/lst_plugin_com.spr.listening.plugin.impl.plugins.SemantriaEnrichmentPlugin/ids*
>
> * 0x34a175e1d5d0130*
>
>
> "0x34a175e1d5d0130" was the ephemeral node session Id. I went back to the
> zookeeper shell and checked the consumers registered for this topic and
> consumer group(same as topic name). Attaching the output in zkCommands.txt.
> This clearly shows that
>
> 10.0.8.222 has ephemeralOwner = 0x34a175e1d5d0130
>
> 10.0.8.225 has ephemeralOwner = 0x34a175e1d5d0127
>
>
> I think we have the issue here that both consumers have written to
> different ephemeral nodes. Watchers are registered for the one of the 2
> ephemeral node. The root cause seems to be the inconsistent state while
> writing the ephemeral nodes in ZK.
>
> Let me know if you need more details.
>
> -Thanks,
>
> Mohit
>
>
>
>
> On Mon, Nov 10, 2014 at 8:46 AM, Neha Narkhede 
> wrote:
>
>> A rebalance should trigger on all consumers when you add a new consumer to
>> the group. If you don't see the zookeeper watch fire, the consumer may
>> have
>> somehow lost the watch. We have seen this behavior on older zk versions, I
>> wonder it that bug got reintroduced. To verify if this is the case, you
>> can
>> run the wchp zookeeper command on the zk leader and check if each consumer
>> has a watch registered.
>>
>> Do you have a way to try this on zk 3.3.4? I would recommend you try the
>> wchp suggestion as well.
>>
>> On Fri, Nov 7, 2014 at 6:07 AM, Mohit Kathuria 
>> wrote:
>>
>> > Hi all,
>> >
>> > Can someone help here. We are getting constant rebalance failure each
>> time
>> > a consumer is added beyond a certain number. Did quite a lot of
>> debugging
>> > on this and still not able to figure out the pattern.
>> >
>> > -Thanks,
>> > Mohit
>> >
>> > On Mon, Nov 3, 2014 at 10:53 PM, Mohit Kathuria > >
>> > wrote:
>> >
>> > > Neha,
>> > >
>> > > Looks like an issue with the consumer rebalance not able to complete
>> > > successfully. We were able to reproduce the issue on topic with 30
>> > > partitions,  3 consumer processes(p1,p2 and p3), properties -  40
>> > > rebalance.max.retries and 1(10s) rebalance.backoff.ms.
>> > >
>> > > Before the process p3 was started, partition ownership was as
>> expected:
>> > >
>> > > partitions 0-14 owned by p1
>> > > partitions 15-29 -> owner p2
>> > >
>> > > As the process p3 started, rebalance was triggered. Process p3 was
>> > > successfully able to acquire partition ownership for partitions 20-29
>> as
>> > > expected as per the rebalance algorithm. However, process p2 while
>> trying
>> > > to acquire ownership of partitions 10-19 saw rebalance failure after
>> 40
>> > > retrie

Re: leader and isr were not set when create the topic

[Neha Narkhede]

There is possibly some error in your broker logs. Can you check if you see
any and send it around?

On Mon, Dec 22, 2014 at 2:46 PM, Sa Li  wrote:

> Hi, All
>
> I created a topic with 3 replications and 6 partitions, but when I check
> this topic, seems there is no leader and isr were set for this topic, see
>
> bin/kafka-topics.sh --create --zookeeper 10.100.98.100:2181
> --replication-factor 3 --partitions 6 --topic perf_producer_p6_test
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> Created topic "perf_producer_p6_test".
>
> root@precise64:/etc/kafka# bin/kafka-topics.sh --describe --zookeeper
> 10.100.98.100:2181 --topic perf_producer_p6_test
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> Topic:perf_producer_p6_test PartitionCount:6
> ReplicationFactor:3 Configs:
> Topic: perf_producer_p6_testPartition: 0Leader: none
> Replicas: 100,101,102   Isr:
> Topic: perf_producer_p6_testPartition: 1Leader: none
> Replicas: 101,102,100   Isr:
> Topic: perf_producer_p6_testPartition: 2Leader: none
> Replicas: 102,100,101   Isr:
> Topic: perf_producer_p6_testPartition: 3Leader: none
> Replicas: 100,102,101   Isr:
> Topic: perf_producer_p6_testPartition: 4Leader: none
> Replicas: 101,100,102   Isr:
> Topic: perf_producer_p6_testPartition: 5Leader: none
> Replicas: 102,101,100   Isr:
>
> Is there a way to specifically set leader and isr in command line, it is
> strange when I create the topic with 5 partitions, it has leader and isr:
> root@precise64:/etc/kafka# bin/kafka-topics.sh --describe --zookeeper
> 10.100.98.100:2181 --topic perf_producer_p5_test
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> Topic:perf_producer_p5_test PartitionCount:5
> ReplicationFactor:3 Configs:
> Topic: perf_producer_p5_testPartition: 0Leader: 102
> Replicas: 102,100,101   Isr: 102,100,101
> Topic: perf_producer_p5_testPartition: 1Leader: 102
> Replicas: 100,101,102   Isr: 102,101
> Topic: perf_producer_p5_testPartition: 2Leader: 101
> Replicas: 101,102,100   Isr: 101,102,100
> Topic: perf_producer_p5_testPartition: 3Leader: 102
> Replicas: 102,101,100   Isr: 102,101,100
> Topic: perf_producer_p5_testPartition: 4Leader: 102
> Replicas: 100,102,101   Isr: 102,101
>
>
> Any ideas?
>
> thanks
>
> --
>
> Alec Li
>



-- 
Thanks,
Neha

Re: Kafka consumer session timeouts

[Neha Narkhede]

Terry,

The zookeeper client used by the high level Kafka consumer has a separate
thread that does the heartbeat in the background. So even if it takes long
to process the message, it should not make the consumer's session to time
out or make the consumer rebalance. You may be running into long GC pauses
on your consumer that might be causing the session timeouts.

Thanks,
Neha

On Fri, Dec 19, 2014 at 12:51 PM, Terry Cumaranatunge 
wrote:

> Hi
> I would like to get some feedback on design choices with kafka consumers.
> We have an application that a consumer reads a message and the thread does
> a number of things, including database accesses before a message is
> produced to another topic. The time between consuming and producing the
> message on the thread can take several minutes. Once message is produced to
> new topic, a commit is done to indicate we are done with work on the
> consumer queue message. Auto commit is disabled for this reason.
>
> I'm using the high level consumer and what I'm noticing is that zookeeper
> and kafka sessions timeout because it is taking too long before we do
> anything on consumer queue so kafka ends up rebalancing every time the
> thread goes back to read more from consumer queue and it starts to take a
> long time before a consumer reads a new message after a while.
>
> I can set zookeeper session timeout very high to not make that a problem
> but then i have to adjust the rebalance parameters accordingly and kafka
> won't pickup a new consumer for a while among other side effects.
>
> What are my options to solve this problem? Is there a way to heartbeat to
> kafka and zookeeper to keep both happy? Do i still have these same issues
> if i were to use a simple consumer?
>
> Thanks
>



-- 
Thanks,
Neha

Re: can't produce message in kafka production

[Neha Narkhede]

The producer is complaining that it's socket channel is already closed.
Which makes me think it was closed due to some error that is not present in
your logs. I'd enable DEBUG and see if that shows the cause.

On Thu, Dec 18, 2014 at 4:13 PM, Gwen Shapira  wrote:
>
> Perhaps you have the logs from broker? It may show other errors that
> can help us troubleshoot.
>
> On Thu, Dec 18, 2014 at 4:11 PM, Sa Li  wrote:
> > Thanks, Gwen, I telnet it,
> > root@precise64:/etc/kafka# telnet 10.100.98.100 9092
> > Trying 10.100.98.100...
> > Connected to 10.100.98.100.
> > Escape character is '^]'.
> >
> > seems it connected, and I check with system operation people, netstate
> > should 9092 is listening. I am assuming this is the connection issue,
> since
> > I can run the same command to my dev-cluster with no problem at all,
> which
> > is 10.100.70.128:9092.
> >
> > Just in case, is it possibly caused by other types of issues?
> >
> > thanks
> >
> > Alec
> >
> > On Thu, Dec 18, 2014 at 2:33 PM, Gwen Shapira 
> wrote:
> >>
> >> Looks like you can't connect to: 10.100.98.100:9092
> >>
> >> I'd validate that this is the issue using telnet and then check the
> >> firewall / ipfilters settings.
> >>
> >> On Thu, Dec 18, 2014 at 2:21 PM, Sa Li  wrote:
> >> > Dear all
> >> >
> >> > We just build a kafka production cluster, I can create topics in kafka
> >> > production from another host. But when I am send very simple message
> as
> >> > producer, it generate such errors:
> >> >
> >> > root@precise64:/etc/kafka# bin/kafka-console-producer.sh
> --broker-list
> >> > 10.100.98.100:9092 --topic my-replicated-topic-production
> >> > SLF4J: Class path contains multiple SLF4J bindings.
> >> > SLF4J: Found binding in
> >> >
> >>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >> > SLF4J: Found binding in
> >> >
> >>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >> > SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> >> > explanation.
> >> > SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> >> > my test message 1
> >> > [2014-12-18 21:44:25,830] WARN Failed to send producer request with
> >> > correlation id 2 to broker 101 with data for partitions
> >> > [my-replicated-topic-production,1]
> >> > (kafka.producer.async.DefaultEventHandler)
> >> > java.nio.channels.ClosedChannelException
> >> > at
> kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
> >> > at
> >> kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
> >> > at
> >> >
> >>
> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
> >> > at
> >> >
> >>
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:103)
> >> > at
> >> >
> >>
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
> >> > at
> >> >
> >>
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
> >> > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> >> > at
> >> >
> >>
> kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:102)
> >> > at
> >> >
> kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
> >> > at
> >> >
> kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
> >> > at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
> >> > at kafka.producer.SyncProducer.send(SyncProducer.scala:101)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:256)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> >> > at
> >> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> >> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99

Re: Kafka Topic/Partition Reassignment issue

[Neha Narkhede]

The reassignment tool outputs the original assignment before executing the
next one. If you have that saved, you can initiate another assignment to
make it go to its initial state. That is probably a safer way to fix the
reassignment.

On Wed, Dec 17, 2014 at 3:12 PM, Salman Ahmed 
wrote:
>
> I had an issue where one kafka node was filling up on disk space. I used
> the reassignment script in an incorrect way, overloading large number of
> topics/partition on two target machines, which caused kafka to stop on
> those machines.
>
> I would like to cancel the reassignment process, and restore it to original
> or no further reassignment. Has anyone had to edit znodes and clean up
> reassignment so replicas and ISR are as they were prior to reassignment. If
> so please share the steps.
>
> Thank you
> Sal
>


-- 
Thanks,
Neha

Re: Built-in replication for broker

[Neha Narkhede]

Replication is available in Kafka 0.8.0 onwards.

On Tue, Dec 16, 2014 at 4:31 PM, Haoming Zhang 
wrote:
>
> Dear developers,
>
> In the paper "Kafka: A Distributed Messaging System for Log Processing,
> Jay Kreps, Neha Narkhede, Jun Rao from LinkedIn, at NetDB workshop 2011",
>
> you mentioned this:
> "If a broker goes down, any message stored on it not yet consumed becomes
> unavailable. If the storage system on a broker is permanently damaged, any
> unconsumed message is lost forever. In the future, we plan to add built-in
> replication in Kafka to redundantly store each me ssage on multiple
> brokers."
>
> I'm curious, does the built-in replication has been implemented or it's
> still in progress? Currently, is there anyway to recover the died broker
> and unconsumed messages?
>
> Thanks,
> Haoming
>



-- 
Thanks,
Neha

Re: Number of Consumers Connected

[Neha Narkhede]

In addition to Gwen's suggestion, we actually don't have jmx metrics that
give you a list of actively consuming processes.

On Mon, Dec 15, 2014 at 12:59 PM, Gwen Shapira 
wrote:
>
> Currently you can find the number of consumer groups through ZooKeeper:
>
> connect to ZK and run
> ls /consumers
>
> and count the number of results
>
> On Mon, Dec 15, 2014 at 11:34 AM, nitin sharma
>  wrote:
> > Hi Team,
> >
> > Is it possible to know how many Consumer Group connected to kafka broker
> Ids
> > and as well as how many Instances within a Group are fetching messages
> from
> > Kafka Brokers
> >
> > Regards,
> > Nitin Kumar Sharma.
>


-- 
Thanks,
Neha

Re: kafka async producer takes a lot of cpu

[Neha Narkhede]

The stack trace you showed shouldn't be going beyond the isDebugEnabled()
check if your log4j properties is configured at a higher log level.
Regarding your producer dropping messages, have you checked the produce
metrics ("kafka.network":name={Produce-TotalTimeMs",type="RequestMetrics")
on the broker to see if the bottleneck is on the server or not?

On Sun, Dec 14, 2014 at 9:31 PM, Rajiv Kurian  wrote:
>
> I'll try it and let you guys know. Is there anything that comes to mind
> from these log messages though? Why would there be so many log messages?
> Would you suggest doing something else to find out why things are working
> so poorly? I am worried about making the risky transition to the beta
> producer client and finding out that things don't improve at all. Another
> thing that was peculiar was that only one of my producers (out of 3, each
> on a separate host) were dropping so many messages.
>
> On Sun, Dec 14, 2014 at 4:58 PM, Neha Narkhede  wrote:
>
> > Thanks for reporting the issue, Rajiv. Since we are actively phasing out
> > the old client, it will be very helpful to know what the behavior on the
> > new client is.
> >
> >
> > On Fri, Dec 12, 2014 at 8:12 PM, Rajiv Kurian 
> > wrote:
> > >
> > > I am using the kafka java api async client (the one that wraps the
> Scala
> > > client). It's dropping quite a bit of data due to the queue being full,
> > and
> > > the process ends up taking a lot of cpu too.
> > >
> > > I am posting to a topic with 1024 partitions (across 3 brokers) - maybe
> > the
> > > high number of brokers is one of the issues. Profiling with YourKit
> > showed
> > > 37% of my CPU being spent on
> > kafka.producer.async.ProducerSendThread.run().
> > > So seems like the producer is not able to keep up with my application
> and
> > > starts dropping. When I expand this waterfall in YourKit, I see that
> 23%
> > > total (not out of the 37%) is being spent on logging! Something like
> > this:
> > >
> > >
> > >
> >
> kafka.producer.BrokerPartitionInfo$$anonfun$getBrokerPartitionInfo$2.apply(PartitionMetadata)
> > > ->kafka.producer.BrokerPartitionInfo.debug(Function0)
> > >->kafka.utils.Logging$class.debug(Logging, Function0)
> > >   ->org.apache.log4j.Category.isDebugEnabled()
> > >  ->... (a bunch of other things that finally break down into)
> > >  ->LoggerContext.java:252
> > >
> > >
> >
> ch.qos.logback.classic.spi.TurboFilterList.getTurboFilterChainDecision(Marker,
> > > Logger, Level, String, Object[], Throwable)
> > >
> > > I am not sure what's going on here. When I look at my process log, none
> > of
> > > these messages are actually logged (probably because of the log level).
> > > Further I don't see anything very suspicious on the broker logs. They
> are
> > > at 60-70% cpu.
> > >
> > > I am planning to try the new Java beta producer client, but I am afraid
> > > something deeper is going on here, that might not be solved by
> switching
> > to
> > > the newer client.
> > >
> >
> >
> > --
> > Thanks,
> > Neha
> >
>


-- 
Thanks,
Neha

Re: kafka producer & consumer problem

[Neha Narkhede]

Hi,

For the producer, we have moved to a new producer API (under
org.apache.kafka.clients.producer.KafkaProducer). Please feel free to give
that a spin and report any issues that you see.

I think the consumer issue you reported is being discussed in another
thread and is fixed in 0.8.2. Can you check if you still see the issue in
0.8.2-beta ?

On Sun, Dec 14, 2014 at 10:08 PM, 黄震  wrote:
>
> Hi,
>
>I'm using kafka-0.8.1.1, this is a good log system, and I'm very
> appreciate for your works. But I'm also found some problem:
>
>1. producer：
>
>  kafka.producer.async.DefaultEventHandler.scala handle(): I suggest
> that sendPartitionPerTopicCache should be cleared every batch, since It
> would not well distributed in every partition.
>
>   2. consumer:
>
>   kafka.consumer.SimpleConsumer.scala disconnect(): I suggest that we
> should delete "if(blockingChannel.isConnected)", since I came across this
> problem when switch broken down, and then broker was abnormally shutted
> down:
>
>  2014-12-04 17:12:14,260 [ReplicaFetcherThread-7-8] ERROR
> kafka.server.ReplicaFetcherThread - [ReplicaFetcherThread-7-8], Error in
> fetch Name: FetchRequest; Version: 0; CorrelationId: 1069738; ClientId:
> ReplicaFetcherThread-7-8; ReplicaId: 3; MaxWait: 500 ms; MinBytes: 1 bytes;
> RequestInfo: [t.v.3,0] -> PartitionFetchInfo(0,1048576)
> java.nio.channels.UnresolvedAddressException at
> sun.nio.ch.Net.checkAddress(Net.java:29) at
> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:512) at
> kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at
> kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44) at
> kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57) at
> kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79) at
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:109)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
> at
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:109)
> at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>
>
>
>
>
> --
>
>
> Best Regards,
>
> Allen huang



-- 
Thanks,
Neha

Re: getOffsetsBefore and latest

[Neha Narkhede]

>
> what about getOffsetsBefore using kafka.api.OffsetRequest.LatestTime? am i
> safe to assume this returns me truly the most recent offset + 1 for each
> partition?


That's right.

On Mon, Dec 15, 2014 at 7:18 AM, Koert Kuipers  wrote:
>
> i read in several places that getOffsetsBefore does not necessary returns
> the last offset before the timestamp, because it is basically file based
> (so it works at the granularity of the files kafka produces).
>
> what about getOffsetsBefore using kafka.api.OffsetRequest.LatestTime? am i
> safe to assume this returns me truly the most recent offset + 1 for each
> partition?
>
> thanks! koert
>


-- 
Thanks,
Neha

Re: kafka async producer takes a lot of cpu

[Neha Narkhede]

Thanks for reporting the issue, Rajiv. Since we are actively phasing out
the old client, it will be very helpful to know what the behavior on the
new client is.


On Fri, Dec 12, 2014 at 8:12 PM, Rajiv Kurian  wrote:
>
> I am using the kafka java api async client (the one that wraps the Scala
> client). It's dropping quite a bit of data due to the queue being full, and
> the process ends up taking a lot of cpu too.
>
> I am posting to a topic with 1024 partitions (across 3 brokers) - maybe the
> high number of brokers is one of the issues. Profiling with YourKit showed
> 37% of my CPU being spent on kafka.producer.async.ProducerSendThread.run().
> So seems like the producer is not able to keep up with my application and
> starts dropping. When I expand this waterfall in YourKit, I see that 23%
> total (not out of the 37%) is being spent on logging! Something like this:
>
>
> kafka.producer.BrokerPartitionInfo$$anonfun$getBrokerPartitionInfo$2.apply(PartitionMetadata)
> ->kafka.producer.BrokerPartitionInfo.debug(Function0)
>->kafka.utils.Logging$class.debug(Logging, Function0)
>   ->org.apache.log4j.Category.isDebugEnabled()
>  ->... (a bunch of other things that finally break down into)
>  ->LoggerContext.java:252
>
> ch.qos.logback.classic.spi.TurboFilterList.getTurboFilterChainDecision(Marker,
> Logger, Level, String, Object[], Throwable)
>
> I am not sure what's going on here. When I look at my process log, none of
> these messages are actually logged (probably because of the log level).
> Further I don't see anything very suspicious on the broker logs. They are
> at 60-70% cpu.
>
> I am planning to try the new Java beta producer client, but I am afraid
> something deeper is going on here, that might not be solved by switching to
> the newer client.
>


-- 
Thanks,
Neha

Re: How to Setup MirrorMaker in Generalized way

[Neha Narkhede]

>
> Is there any solution to avoid this duplicated entry in target cluster? I
> am using Kafka


That is the expected behavior when the mirror maker process is killed. It
is expected that in most cases, it be shutdown cleanly. If it is killed, it
doesn't get a chance to checkpoint it's offset, which causes the duplicates.

Thanks,
Neha

On Thu, Dec 11, 2014 at 4:30 AM, Madhukar Bharti 
wrote:
>
> Hi Neha,
>
> Thanks for your reply.
>
> Now using MM tool to replicate data between Kafka clusters, But I am facing
> one problem, Messages gets duplicated if MM killed forcefully[ *kill -9* ].
>
> Is there any solution to avoid this duplicated entry in target cluster? I
> am using Kafka
>
>
> *8.1.1.*
>
> On Mon, Dec 8, 2014 at 11:17 PM, Neha Narkhede  wrote:
>
> > Hi Madhukar,
> >
> > From the same documentation link you referred to -
> >
> > The source and destination clusters are completely independent entities:
> > > they can have different numbers of partitions and the offsets will not
> be
> > > the same. For this reason the mirror cluster is not really intended as
> a
> > > fault-tolerance mechanism (as the consumer position will be different);
> > for
> > > that we recommend using normal in-cluster replication. The mirror maker
> > > process will, however, retain and use the message key for partitioning
> so
> > > order is preserved on a per-key basis.
> >
> >
> > There is no way to setup an *exact* Kafka mirror yet.
> >
> > Thanks,
> > Neha
> >
> > On Mon, Dec 8, 2014 at 7:47 AM, Madhukar Bharti <
> bhartimadhu...@gmail.com>
> > wrote:
> >
> > > Hi,
> > >
> > > I am going to setup Kafka clusters having 3 brokers in Datacenter 1.
> > Topics
> > > can be created time to time. Each topic can have varying partitions
> > mostly
> > > 1,10 or 20. Each application might have different partitioning
> algorithm
> > > that we don't know(let it be hidden from ops team).
> > >
> > > We want to setup mirror maker tool in such a way so that, the exact
> > > partitioned data should go to the same partition without knowing the
> > Topics
> > > partition logic and it should be *generalized*. [This should be common
> > for
> > > all Topics.]
> > >
> > > *like partition 0 at DataCenter1 should be exact mirror of  partition-0
> > in
> > > Datacenter2*.
> > >
> > > Please suggest me a solution for doing so. If MirrorMaker
> > > <http://kafka.apache.org/documentation.html#basic_ops_mirror_maker>
> tool
> > > provide any configurations which solve this use-case please let me
> know.
> > >
> > >
> > >
> > > Regards,
> > > Madhukar Bharti
> > >
> >
> >
> >
> > --
> > Thanks,
> > Neha
> >
>
>
> --
> Thanks and Regards,
> Madhukar Bharti
>


-- 
Thanks,
Neha

Re: How to raise a question in forum

[Neha Narkhede]

You can post it here :-)

On Tue, Dec 9, 2014 at 1:26 PM, nitin sharma 
wrote:

> HI,
> I need some help related to Kafka Consumer Lag monitoring.. Kindly let me
> know how can i post my query to the forum.
>
> Regards,
> Nitin Kumar Sharma.
>



-- 
Thanks,
Neha

Re: Increased CPU usage with 0.8.2-beta

[Neha Narkhede]

The following should be sufficient

java
-agentlib:hprof=cpu=samples,depth=100,interval=20,lineno=y,thread=y,file=kafka.hprof


You would need to start the Kafka server with the settings above for
sometime until you observe the problem.

On Tue, Dec 9, 2014 at 3:47 AM, Mathias Söderberg <
mathias.soederb...@gmail.com> wrote:

> Hi Neha,
>
> Yeah sure. I'm not familiar with hprof, so any particular options I should
> include or just run with defaults?
>
> Best regards,
> Mathias
>
> On Mon Dec 08 2014 at 7:41:32 PM Neha Narkhede  wrote:
>
> > Thanks for reporting the issue. Would you mind running hprof and sending
> > the output?
> >
> > On Mon, Dec 8, 2014 at 1:25 AM, Mathias Söderberg <
> > mathias.soederb...@gmail.com> wrote:
> >
> > > Good day,
> > >
> > > I upgraded a Kafka cluster from v0.8.1.1 to v0.8.2-beta and noticed
> that
> > > the CPU usage on the broker machines went up by roughly 40%, from ~60%
> to
> > > ~100% and am wondering if anyone else has experienced something
> similar?
> > > The load average also went up by 2x-3x.
> > >
> > > We're running on EC2 and the cluster currently consists of four
> > m1.xlarge,
> > > with roughly 1100 topics / 4000 partitions. Using Java 7 (1.7.0_65 to
> be
> > > exact) and Scala 2.9.2. Configurations can be found over here:
> > > https://gist.github.com/mthssdrbrg/7df34a795e07eef10262.
> > >
> > > I'm assuming that this is not expected behaviour for 0.8.2-beta?
> > >
> > > Best regards,
> > > Mathias
> > >
> >
> >
> >
> > --
> > Thanks,
> > Neha
> >
>



-- 
Thanks,
Neha

Re: Topic config reset on alter

[Neha Narkhede]

Thanks for reporting the bug. Would you mind filing a JIRA

?

On Tue, Dec 9, 2014 at 12:46 PM, Andrew Jorgensen <
ajorgen...@twitter.com.invalid> wrote:

> I am using kafka 8.1.1
>
> If you run an alter on a topic to increase the number of partitions the
> existing configs are removed from the topic. You can confirm this by doing:
>
> bin/kafka-topics.sh —zookeeper localhost —alter —topic topic —config
> retention.ms=360
>
> bin/kafka-topics.sh —zookeeper localhost —describe
> > Topic:topic  PartitionCount:5   ReplicationFactor:1 Configs:
> retention.ms=360
>
> bin/kafka-topics.sh —zookeeper localhost —alter —topic topic —partitions 10
>
> bin/kafka-topics.sh —zookeeper localhost —describe
> > Topic:topic  PartitionCount:10   ReplicationFactor:1 Configs:
>
> This feels like a bug to me, where is the best place to report something
> like this?
> --
> Andrew Jorgensen
> @ajorgensen




-- 
Thanks,
Neha

Re: Reading only the latest message

[Neha Narkhede]

>
> Is this the best way to get the offset and is it safe to decrement the
> offset returned as we do in the sample code below.



Yes, this should work. Let us know if you see any problems.

On Tue, Dec 9, 2014 at 7:44 AM, Orelowitz, David 
wrote:

>  I am reposting a question that I posted last week.
>
>
>
> On startup or recovery we would like to read the latest message in each
> partition.
>
> The getOffsetsBefore() below seems to return the offset of the next
> message that will be published to that partition.
>
>
>
> The code below works correctly as required.
>
> Is this the best way to get the offset and is it safe to decrement the
> offset returned as we do in the sample code below.
>
>
>
> requestInfo.put(topicAndPartition, *new* PartitionOffsetRequestInfo(-1,
> 1));
>
> kafka.javaapi.OffsetRequest request = *new* kafka.javaapi.OffsetRequest(
> requestInfo, kafka.api.OffsetRequest.*CurrentVersion*(), clientName);
>
> OffsetResponse response = consumer.getOffsetsBefore(request);
>
>
>
> *if*(readOffset != 0)
>
> readOffset--;
>
> else
>
>handle this case…
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> *From:* Neha Narkhede [mailto:n...@confluent.io]
> *Sent:* Monday, December 08, 2014 12:43 PM
> *To:* Orelowitz, David
> *Cc:* users@kafka.apache.org
>
> *Subject:* Re: Reading only the latest message
>
>
>
> The returned latest offset - 1 will be the offset of the last message.
> Sorry, should've made it clear in my last email. Let me know if that helps.
>
>
>
> On Mon, Dec 8, 2014 at 8:32 AM, Orelowitz, David 
> wrote:
>
> Neha,
>
> This seems to return the offset of the next message that will be
> published. If I fetch at that offset I will block until a new message is
> published to that partition.
>
> I am actually trying to read the contents of the latest message in the
> partition, and based on info in the message resubscribe to the data source.
>
>
> -Original Message-
> From: Neha Narkhede [mailto:n...@confluent.io]
> Sent: Friday, December 05, 2014 8:33 PM
> To: users@kafka.apache.org
> Subject: Re: Reading only the latest message
>
> You can use the getOffsetsBefore() API and specify -1L to get the offset
> of the last committed message (at the time of the request) for that
> partition.
>
> On Fri, Dec 5, 2014 at 12:42 PM, Orelowitz, David <
> david.orelow...@baml.com>
> wrote:
>
> > What is the best mechanism to retrieve the latest message from a kafka
> > partition.
> >
> > We intend for our producer, on startup or recovery, to read the
> > upstream sequence number in the last message in the partition and
> > request for the upstream system to start sending from that sequence
> number++.
> >
> > Currently we are creating a SimpleConsumer and then calling
> > getOffsetBefore() using the current wall time. We then decrement the
> > offset returned and retrieve the message at this offset. We do manage
> > the case when the offset is zero.
> >
> > It seem to work!
> >
> > Is this the right approach.
> >
> > Thanks,
> > David
> >
>
> --
> Thanks,
> Neha
>
> --
> This message, and any attachments, is for the intended recipient(s) only,
> may contain information that is privileged, confidential and/or proprietary
> and subject to important terms and conditions available at
> http://www.bankofamerica.com/emaildisclaimer.   If you are not the
> intended recipient, please delete this message.
>
>
>
>
>
> --
>
> Thanks,
> Neha
>   --
> This message, and any attachments, is for the intended recipient(s) only,
> may contain information that is privileged, confidential and/or proprietary
> and subject to important terms and conditions available at
> http://www.bankofamerica.com/emaildisclaimer. If you are not the intended
> recipient, please delete this message.
>



-- 
Thanks,
Neha

Re: Increased CPU usage with 0.8.2-beta

[Neha Narkhede]

Thanks for reporting the issue. Would you mind running hprof and sending
the output?

On Mon, Dec 8, 2014 at 1:25 AM, Mathias Söderberg <
mathias.soederb...@gmail.com> wrote:

> Good day,
>
> I upgraded a Kafka cluster from v0.8.1.1 to v0.8.2-beta and noticed that
> the CPU usage on the broker machines went up by roughly 40%, from ~60% to
> ~100% and am wondering if anyone else has experienced something similar?
> The load average also went up by 2x-3x.
>
> We're running on EC2 and the cluster currently consists of four m1.xlarge,
> with roughly 1100 topics / 4000 partitions. Using Java 7 (1.7.0_65 to be
> exact) and Scala 2.9.2. Configurations can be found over here:
> https://gist.github.com/mthssdrbrg/7df34a795e07eef10262.
>
> I'm assuming that this is not expected behaviour for 0.8.2-beta?
>
> Best regards,
> Mathias
>



-- 
Thanks,
Neha

Re: How to Setup MirrorMaker in Generalized way

[Neha Narkhede]

Hi Madhukar,

>From the same documentation link you referred to -

The source and destination clusters are completely independent entities:
> they can have different numbers of partitions and the offsets will not be
> the same. For this reason the mirror cluster is not really intended as a
> fault-tolerance mechanism (as the consumer position will be different); for
> that we recommend using normal in-cluster replication. The mirror maker
> process will, however, retain and use the message key for partitioning so
> order is preserved on a per-key basis.


There is no way to setup an *exact* Kafka mirror yet.

Thanks,
Neha

On Mon, Dec 8, 2014 at 7:47 AM, Madhukar Bharti 
wrote:

> Hi,
>
> I am going to setup Kafka clusters having 3 brokers in Datacenter 1. Topics
> can be created time to time. Each topic can have varying partitions mostly
> 1,10 or 20. Each application might have different partitioning algorithm
> that we don't know(let it be hidden from ops team).
>
> We want to setup mirror maker tool in such a way so that, the exact
> partitioned data should go to the same partition without knowing the Topics
> partition logic and it should be *generalized*. [This should be common for
> all Topics.]
>
> *like partition 0 at DataCenter1 should be exact mirror of  partition-0 in
> Datacenter2*.
>
> Please suggest me a solution for doing so. If MirrorMaker
>  tool
> provide any configurations which solve this use-case please let me know.
>
>
>
> Regards,
> Madhukar Bharti
>



-- 
Thanks,
Neha

Re: Reading only the latest message

[Neha Narkhede]

The returned latest offset - 1 will be the offset of the last message.
Sorry, should've made it clear in my last email. Let me know if that helps.

On Mon, Dec 8, 2014 at 8:32 AM, Orelowitz, David 
wrote:

> Neha,
>
> This seems to return the offset of the next message that will be
> published. If I fetch at that offset I will block until a new message is
> published to that partition.
>
> I am actually trying to read the contents of the latest message in the
> partition, and based on info in the message resubscribe to the data source.
>
> -Original Message-
> From: Neha Narkhede [mailto:n...@confluent.io]
> Sent: Friday, December 05, 2014 8:33 PM
> To: users@kafka.apache.org
> Subject: Re: Reading only the latest message
>
> You can use the getOffsetsBefore() API and specify -1L to get the offset
> of the last committed message (at the time of the request) for that
> partition.
>
> On Fri, Dec 5, 2014 at 12:42 PM, Orelowitz, David <
> david.orelow...@baml.com>
> wrote:
>
> > What is the best mechanism to retrieve the latest message from a kafka
> > partition.
> >
> > We intend for our producer, on startup or recovery, to read the
> > upstream sequence number in the last message in the partition and
> > request for the upstream system to start sending from that sequence
> number++.
> >
> > Currently we are creating a SimpleConsumer and then calling
> > getOffsetBefore() using the current wall time. We then decrement the
> > offset returned and retrieve the message at this offset. We do manage
> > the case when the offset is zero.
> >
> > It seem to work!
> >
> > Is this the right approach.
> >
> > Thanks,
> > David
> >
> >
> > --
> > This message, and any attachments, is for the intended recipient(s)
> > only, may contain information that is privileged, confidential and/or
> > proprietary and subject to important terms and conditions available at
> > http://www.bankofamerica.com/emaildisclaimer.   If you are not the
> > intended recipient, please delete this message.
> >
>
>
>
> --
> Thanks,
> Neha
>
> --
> This message, and any attachments, is for the intended recipient(s) only,
> may contain information that is privileged, confidential and/or proprietary
> and subject to important terms and conditions available at
> http://www.bankofamerica.com/emaildisclaimer.   If you are not the
> intended recipient, please delete this message.
>



-- 
Thanks,
Neha

Re: Broker don't get back when killed and restarted

[Neha Narkhede]

Have you tried using the latest stable version of Kafka (0.8.1.1) with
controlled shutdown?

On Fri, Dec 5, 2014 at 2:39 PM, Haeley Yao  wrote:

> Hi, Kafka group
>
> We try to improve the fault-tolerance of kafka cluster. We setup 4 nodes
> kafka cluster and 3 nodes zookeeper cluster.
>
> ubuntu version: Ubuntu 14.04.1
> zookeeper version: 3.4.5-1392090, built on 09/30/2012 17:52 GMT
> kafka version: kafka_2.8.0-0.8.0
>
> kafka0.x.x.x
> broker:9092
> borker.id = 11
> zookeeper.connect=zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182
>
> kafka1.x.x.x
> broker:9092
> borker.id = 12
> zookeeper.connect=zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182
>
> kafka2.x.x.x
> broker:9092
> borker.id = 13
> zookeeper.connect=zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182
>
> kafka3.x.x.x
> broker:9092
> borker.id = 14
> zookeeper.connect=zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182
>
> 1. start both kafka and zk servers, everything OK
>
> 2. create 2 topics
> bin/kafka-create-topic.sh --zookeeper
> zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182 —parition 3 --replica 3
> --topic zerg.hydra
>
> bin/kafka-create-topic.sh --zookeeper
> zk-01.x.x.x:2182,zk-02.x.x.x:2182,zk-03.x.x.x:2182 —parition 2 --replica 3
> --topic zerg.test
>
> 3. start producer and consumer on brokerId 11. the inputs sent to producer
> are received by consumer.
>
> 4. stop kafka servers on brokerId 12, 13, 14 by supervisor
> /etc/init.d/supervisor stop
>
> only brokerId 11 is running.
> root@kafka0:/home/dev/kafka# bin/kafka-list-topic.sh --topic zerg.hydra
> --zookeeper zk-01.dev.quantifind.com:2182,zk-02.dev.quantifind.com:2182,
> zk-03.dev.quantifind.com:2182
> topic: zerg.hydra   partition: 0leader: 11  replicas:
> 11,14,12  isr: 11
> topic: zerg.hydra   partition: 1leader: 11  replicas:
> 11,14,12  isr: 11
> topic: zerg.hydra   partition: 2leader: 11  replicas:
> 12,11,13  isr: 11
>
> root@kafka0:/home/dev/kafka# bin/kafka-list-topic.sh --topic zerg.test
> --zookeeper zk-01.dev.quantifind.com:2182,zk-02.dev.quantifind.com:2182,
> zk-03.dev.quantifind.com:2182
> topic: zerg.testpartition: 0leader: 11  replicas:
> 13,14,11  isr: 11
> topic: zerg.testpartition: 1leader: 11  replicas:
> 14,11,12  isr: 11
>
> 5. start kafka on brokerId 13, kafka2.x.x.x
> bin/kafka-server-start.sh config/server.properties
>
> [2014-12-05 14:34:45,607] ERROR [KafkaApi-13] error when handling request
> Name: FetchRequest; Version: 0; CorrelationId: 222; ClientId:
> ReplicaFetcherThread-0-11; ReplicaId: 13; MaxWait: 500 ms; MinBytes: 1
> bytes; RequestInfo: [zerg.hydra,2] ->
> PartitionFetchInfo(0,1048576),[zerg.test,0] ->
> PartitionFetchInfo(15,1048576) (kafka.server.KafkaApis)
> kafka.common.KafkaException: Shouldn't set logEndOffset for replica 13
> partition [zerg.hydra,2] since it's local
> at kafka.cluster.Replica.logEndOffset_$eq(Replica.scala:46)
> at
> kafka.cluster.Partition.updateLeaderHWAndMaybeExpandIsr(Partition.scala:227)
>
> Could you help on it?
>
> Thank you!
>
>
>
>
> Haeley
> —
> Work hard, stay humble.
>
>
>
>
>


-- 
Thanks,
Neha

Re: Reading only the latest message

[Neha Narkhede]

You can use the getOffsetsBefore() API and specify -1L to get the offset of
the last committed message (at the time of the request) for that partition.

On Fri, Dec 5, 2014 at 12:42 PM, Orelowitz, David 
wrote:

> What is the best mechanism to retrieve the latest message from a kafka
> partition.
>
> We intend for our producer, on startup or recovery, to read the upstream
> sequence number in the last message in the partition and request for the
> upstream system to start sending from that sequence number++.
>
> Currently we are creating a SimpleConsumer and then calling
> getOffsetBefore() using the current wall time. We then decrement the offset
> returned and retrieve the message at this offset. We do manage the case
> when the offset is zero.
>
> It seem to work!
>
> Is this the right approach.
>
> Thanks,
> David
>
>
> --
> This message, and any attachments, is for the intended recipient(s) only,
> may contain information that is privileged, confidential and/or proprietary
> and subject to important terms and conditions available at
> http://www.bankofamerica.com/emaildisclaimer.   If you are not the
> intended recipient, please delete this message.
>



-- 
Thanks,
Neha

Re: Can Mirroring Preserve Every Topic's Partition?

[Neha Narkhede]

Going back to your previous requirement of ensuring that the data in the
target cluster is in the same order as the source cluster, all you need is
to specify a key with every record in your data. The mirror maker and its
producer takes care of placing all the data for a particular key in the
same partition on the target cluster. Effectively, all your data will be in
the same order (though there may be a few duplicates as I mentioned before).

Hope that helps!

On Fri, Dec 5, 2014 at 1:23 PM, Alex Melville  wrote:

> Thank you for your replies Guozhang and Neha, though I have some followup
> questions.
>
> I wrote my own Java Consumer and Producer based off of the Kafka Producer
> API and High Level Consumer. Let's call them MyConsumer and MyProducer.
> MyProducer uses a custom Partitioner class called SimplePartitioner. In the
> producer.config file that I specify when I run the MirrorMaker from the
> command line, there is a parameter "partitioner.class". I keep getting
> "ClassDefNotFoundException exceptions, no matter if I put the absolute path
> to my SimplePartitioner.class file, a relative path, or even when I add
> SimplePartitioner.class to the $CLASSPATH variables created in the
> kafka-run-class.sh script. Here is my output error:
>
> Exception in thread "main" java.lang.ClassNotFoundException:
> SimplePartitioner.class
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:191)
> at kafka.utils.Utils$.createObject(Utils.scala:438)
> at kafka.producer.Producer.(Producer.scala:60)
> at kafka.tools.MirrorMaker$$anonfun$1.apply(MirrorMaker.scala:116)
> at kafka.tools.MirrorMaker$$anonfun$1.apply(MirrorMaker.scala:106)
> at
>
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
> at
>
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
> at scala.collection.immutable.Range.foreach(Range.scala:81)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
> at scala.collection.immutable.Range.map(Range.scala:46)
> at kafka.tools.MirrorMaker$.main(MirrorMaker.scala:106)
> at kafka.tools.MirrorMaker.main(MirrorMaker.scala)
>
> What is the correct value for the "partitioner.class" parameter in my
> producer.properties config file?
>
>
>
>
> Guozhang, in your reply to my original message you said "When the consumer
> of the MM gets a message, put the message to the producer's queue...". This
> seems to imply that I can specify my own custom Consumer and Producer when
> I run the Mirrormaker. How can I do this? Or, if I'm understand incorrectly
> and I have to use whichever default consumer/producer the Mirrormaker uses,
> how can I get that consumer to learn which partition it's reading from,
> pass that info to the producer, and then specify that partition ID when the
> producer rights to the target cluster?
>
>
> -Alex
>
> On Wed, Nov 26, 2014 at 10:33 AM, Guozhang Wang 
> wrote:
>
> > Hello Alex,
> >
> > This can be done by doing some tweaks in the MM code (with the 0.8.2 new
> > producer).
> >
> > 1. Set-up your MM to have the total # of producers equal to the #. of
> > partitions in source / target cluster.
> >
> > 2. When the consumer of the MM gets a message, put the message to the
> > producer's queue based on its partition id; i.e. if the partition id is
> n,
> > put to n's producer queue.
> >
> > 3. When producer sends the data, specify the partition id; so each
> producer
> > will only send to a single partition.
> >
> > Guozhang
> >
> >
> > On Tue, Nov 25, 2014 at 8:19 PM, Alex Melville 
> > wrote:
> >
> > > Howdy friends,
> > >
> > >
> > > I'd like to mirror the topics on several clusters to a central cluster,
> > and
> > > I'm looking at using the default Mirrormaker to do so. I've already
> done
> > > some basic testing on the Mirrormaker found here:
> > >
> > >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330
> > >
> > > and managed to successfully copy a topic's partitions on a source
> cluster
> > > to a topic on a target cluster. So I'm able to mirror correctly.
> However
> > > for my particular use case I need to ensure that when I copy a topic's
> > > partitions from source cluster to target cluster, a partition created
> on
> > > the target cluster contains data in the exact same order as the data on
> > the
> > > corresponding partition on the source cluster.
> > >
> > > I'm thinking of writing a Simple Consumer so I can manually compare the
> > > events in a source cluster's partition with the corresponding partition
> >

Re: kafka consumer to write into DB

[Neha Narkhede]

Not that I know of.

On Fri, Dec 5, 2014 at 9:44 AM, Sa Li  wrote:

> Thanks, Neha, is there a java version batch consumer?
>
> thanks
>
>
>
> On Fri, Dec 5, 2014 at 9:41 AM, Scott Clasen  wrote:
>
> > if you are using scala/akka this will handle the batching and acks for
> you.
> >
> > https://github.com/sclasen/akka-kafka#akkabatchconsumer
> >
> > On Fri, Dec 5, 2014 at 9:21 AM, Sa Li  wrote:
> >
> > > Thank you very much for the reply, Neha, I have a question about
> > consumer,
> > > I consume the data from kafka and write into DB, of course I have to
> > create
> > > a hash map in memory, load data into memory and bulk copy to DB instead
> > of
> > > insert into DB line by line. Does it mean I need to ack each message
> > while
> > > load to memory?
> > >
> > > thanks
> > >
> > >
> > >
> > > On Thu, Dec 4, 2014 at 1:21 PM, Neha Narkhede 
> wrote:
> > >
> > > > This is specific for pentaho but may be useful -
> > > > https://github.com/RuckusWirelessIL/pentaho-kafka-consumer
> > > >
> > > > On Thu, Dec 4, 2014 at 12:58 PM, Sa Li  wrote:
> > > >
> > > > > Hello, all
> > > > >
> > > > > I never developed a kafka consumer, I want to be able to make an
> > > advanced
> > > > > kafka consumer in java to consume the data and continuously write
> the
> > > > data
> > > > > into postgresql DB. I am thinking to create a map in memory and
> > > getting a
> > > > > predefined number of messages in memory then write into DB in
> batch,
> > is
> > > > > there a API or sample code to allow me to do this?
> > > > >
> > > > >
> > > > > thanks
> > > > >
> > > > >
> > > > > --
> > > > >
> > > > > Alec Li
> > > > >
> > > >
> > > >
> > > >
> > > > --
> > > > Thanks,
> > > > Neha
> > > >
> > >
> > >
> > >
> > > --
> > >
> > > Alec Li
> > >
> >
>
>
>
> --
>
> Alec Li
>



-- 
Thanks,
Neha

Re: kafka consumer to write into DB

[Neha Narkhede]

This is specific for pentaho but may be useful -
https://github.com/RuckusWirelessIL/pentaho-kafka-consumer

On Thu, Dec 4, 2014 at 12:58 PM, Sa Li  wrote:

> Hello, all
>
> I never developed a kafka consumer, I want to be able to make an advanced
> kafka consumer in java to consume the data and continuously write the data
> into postgresql DB. I am thinking to create a map in memory and getting a
> predefined number of messages in memory then write into DB in batch, is
> there a API or sample code to allow me to do this?
>
>
> thanks
>
>
> --
>
> Alec Li
>



-- 
Thanks,
Neha

Re: Can Mirroring Preserve Every Topic's Partition?

[Neha Narkhede]

As Guozhang mentioned, that will allow exact partitioning but note that
this still doesn't ensure identical partitions since your target cluster
can have duplicates depending on whether the producer retried sending data
or not.

On Wed, Nov 26, 2014 at 10:33 AM, Guozhang Wang  wrote:

> Hello Alex,
>
> This can be done by doing some tweaks in the MM code (with the 0.8.2 new
> producer).
>
> 1. Set-up your MM to have the total # of producers equal to the #. of
> partitions in source / target cluster.
>
> 2. When the consumer of the MM gets a message, put the message to the
> producer's queue based on its partition id; i.e. if the partition id is n,
> put to n's producer queue.
>
> 3. When producer sends the data, specify the partition id; so each producer
> will only send to a single partition.
>
> Guozhang
>
>
> On Tue, Nov 25, 2014 at 8:19 PM, Alex Melville 
> wrote:
>
> > Howdy friends,
> >
> >
> > I'd like to mirror the topics on several clusters to a central cluster,
> and
> > I'm looking at using the default Mirrormaker to do so. I've already done
> > some basic testing on the Mirrormaker found here:
> >
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330
> >
> > and managed to successfully copy a topic's partitions on a source cluster
> > to a topic on a target cluster. So I'm able to mirror correctly. However
> > for my particular use case I need to ensure that when I copy a topic's
> > partitions from source cluster to target cluster, a partition created on
> > the target cluster contains data in the exact same order as the data on
> the
> > corresponding partition on the source cluster.
> >
> > I'm thinking of writing a Simple Consumer so I can manually compare the
> > events in a source cluster's partition with the corresponding partition
> on
> > the target cluster, but I'm not 100% sure if I'll be able to verify my
> > guarantee if I do it this way. Can anyone here verify that partitions
> > copied over to the target cluster by the default Mirrormaker are an exact
> > copy of those on the source cluster?
> >
> >
> > Thanks in advance,
> >
> > Alex Melville
> >
>
>
>
> --
> -- Guozhang
>

Re: Best practice for upgrading Kafka cluster from 0.8.1 to 0.8.1.1

[Neha Narkhede]

Will doing one broker at
a time by brining the broker down, updating the code, and restarting it be
sufficient?

Yes this should work for the upgrade.

On Mon, Dec 1, 2014 at 10:23 PM, Yu Yang  wrote:

> Hi,
>
> We have a kafka cluster that runs Kafka 0.8.1 that we are considering
> upgrade to 0.8.1.1. The Kafka documentation
>  mentions upgrading
> from 0.8 to 0.8.1, but not from 0.8.1 to 0.8.1.1.  Will doing one broker at
> a time by brining the broker down, updating the code, and restarting it be
> sufficient? Any best practice suggestions?
>
> Thanks!
>
> Regards,
> Yu
>

Re: Questions about new consumer API

[Neha Narkhede]

The offsets are keyed on  so if you have more than
one owner per partition, they will rewrite each other's offsets and lead to
incorrect state.

On Tue, Dec 2, 2014 at 2:32 PM, hsy...@gmail.com  wrote:

> Thanks Neha, another question, so if offsets are stored under group.id,
> dose it mean in one group, there should be at most one subscriber for each
> topic partition?
>
> Best,
> Siyuan
>
> On Tue, Dec 2, 2014 at 12:55 PM, Neha Narkhede 
> wrote:
>
> > 1. In this doc it says kafka consumer will automatically do load balance.
> > Is it based on throughtput or same as what we have now balance the
> > cardinality among all consumers in same ConsumerGroup? In a real case
> > different partitions could have different peak time.
> >
> > Load balancing is still based on # of partitions for the subscribed
> topics
> > and
> > ensuring that each partition has exactly one consumer as the owner.
> >
> > 2. In the API, threre is subscribe(partition...) method saying not using
> > group management, does it mean the group.id property will be discarded
> and
> > developer has full control of distributing partitions to consumers?
> >
> > group.id is also required for offset management, if the user chooses to
> > use
> > Kafka based offset management. The user will have full control over
> > distribution
> > of partitions to consumers.
> >
> > 3. Is new API compatible with old broker?
> >
> > Yes, it will.
> >
> > 4. Will simple consumer api and high-level consumer api still be
> supported?
> >
> > Over time, we will phase out the current high-level and simple consumer
> > since the
> > 0.9 API supports both.
> >
> > Thanks,
> > Neha
> >
> > On Tue, Dec 2, 2014 at 12:07 PM, hsy...@gmail.com 
> > wrote:
> >
> > > Hi guys,
> > >
> > > I'm interested in the new Consumer API.
> > > http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/
> > >
> > > I have couple of question.
> > > 1. In this doc it says kafka consumer will automatically do load
> balance.
> > > Is it based on throughtput or same as what we have now balance the
> > > cardinality among all consumers in same ConsumerGroup? In a real case
> > > different partitions could have different peak time.
> > > 2. In the API, threre is subscribe(partition...) method saying not
> using
> > > group management, does it mean the group.id property will be discarded
> > and
> > > developer has full control of distributing partitions to consumers?
> > > 3. Is new API compatible with old broker?
> > > 4. Will simple consumer api and high-level consumer api still be
> > supported?
> > >
> > > Thanks!
> > >
> > > Best,
> > > Siyuan
> > >
> >
>

Re: Questions about new consumer API

[Neha Narkhede]

1. In this doc it says kafka consumer will automatically do load balance.
Is it based on throughtput or same as what we have now balance the
cardinality among all consumers in same ConsumerGroup? In a real case
different partitions could have different peak time.

Load balancing is still based on # of partitions for the subscribed topics
and
ensuring that each partition has exactly one consumer as the owner.

2. In the API, threre is subscribe(partition...) method saying not using
group management, does it mean the group.id property will be discarded and
developer has full control of distributing partitions to consumers?

group.id is also required for offset management, if the user chooses to use
Kafka based offset management. The user will have full control over
distribution
of partitions to consumers.

3. Is new API compatible with old broker?

Yes, it will.

4. Will simple consumer api and high-level consumer api still be supported?

Over time, we will phase out the current high-level and simple consumer
since the
0.9 API supports both.

Thanks,
Neha

On Tue, Dec 2, 2014 at 12:07 PM, hsy...@gmail.com  wrote:

> Hi guys,
>
> I'm interested in the new Consumer API.
> http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/
>
> I have couple of question.
> 1. In this doc it says kafka consumer will automatically do load balance.
> Is it based on throughtput or same as what we have now balance the
> cardinality among all consumers in same ConsumerGroup? In a real case
> different partitions could have different peak time.
> 2. In the API, threre is subscribe(partition...) method saying not using
> group management, does it mean the group.id property will be discarded and
> developer has full control of distributing partitions to consumers?
> 3. Is new API compatible with old broker?
> 4. Will simple consumer api and high-level consumer api still be supported?
>
> Thanks!
>
> Best,
> Siyuan
>

Re: Kafka 0.8.2 release - before Santa Claus?

[Neha Narkhede]

+1 for doing a 0.8.2 final before the December break.

On Mon, Dec 1, 2014 at 8:40 AM, Jun Rao  wrote:

> We are currently discussing a last-minute API change to the new java
> producer. We have also accumulated a few more 0.8.2 blockers.
>
>
> https://issues.apache.org/jira/browse/KAFKA-1642?jql=project%20%3D%20KAFKA%20AND%20status%20in%20(Open%2C%20%22In%20Progress%22%2C%20Reopened%2C%20%22Patch%20Available%22)%20AND%20priority%20%3D%20Blocker%20AND%20fixVersion%20%3D%200.8.2
>
> So, we likely will have another 0.8.2 release in Dec. However, I am not
> sure if that's beta2 or final.
>
> Thanks,
>
> Jun
>
> On Wed, Nov 26, 2014 at 12:22 PM, Otis Gospodnetic <
> otis.gospodne...@gmail.com> wrote:
>
> > Hi,
> >
> > People using SPM to monitor Kafka have been anxiously asking us about the
> > 0.8.2 release and we've been telling them December.  Is that still the
> > plan?
> >
> > Thanks,
> > Otis
> > --
> > Monitoring * Alerting * Anomaly Detection * Centralized Log Management
> > Solr & Elasticsearch Support * http://sematext.com/
> >
>

Re: logging agent based on fuse and kafka: first release

[Neha Narkhede]

Great. Thanks for sharing. I added it to our ecosystem
 wiki.

On Tue, Nov 25, 2014 at 9:58 AM, yazgoo  wrote:

> Hi,
>
> First I'd like to thank kafka developers for writing kafka.
>
> This is an announcement for the first release of a file system logging
> agent based on kafka.
>
> It is written for collecting logs from servers running all kind of
> software,
> as a generic way to collect logs without needing to know about each logger.
>
> Home:
> https://github.com/yazgoo/fuse_kafka
>
> Here are some functionnalities:
>
>- sends all writes to given directories to kafka
>- passes through FS syscalls to underlying directory
>- captures the pid, gid, uid, user, group, command line doing the write
>- you can add metadata to identify from where the message comes from
>(e.g. ip-address, ...)
>- you can configure kafka destination cluster either by giving a broker
>list or a zookeeper list
>- you can specify a bandwidth quota: fuse_kafka won't send data if a
>file is written more than a given size per second (useful for preventing
>floods caused by core files dumped or log rotations in directories
> watched
>by fuse_kafka)
>
> It is based on:
>
>- FUSE (filesystem in userspace), to capture writes done under a given
>directory
>- kafka (messaging queue), as the event transport system
>- logstash: events are written to kafka in logstash format (except
>messages and commands which are stored in base64)
>
> It is written in C and python.
>
> Packages are provided for various distros, see installing section in
> README.md.
> FUSE adds an overhead, so it should not be used on filesystems where high
> throughput is necessary.
> Here are benchmarks:
>
> http://htmlpreview.github.io/?https://raw.githubusercontent.com/yazgoo/fuse_kafka/master/benchs/benchmarks.html
>
> Contributions are welcome, of course!
>
> Regards
>

Re: Broker keeps rebalancing

[Neha Narkhede]

@Neha, Can you share suggested consumer side GC settings?

Consumer side GC settings are not standard since it is a function of your
application that embeds the consumer. Your consumer application's memory
patterns will dictate your GC settings. Sorry, I know that's not very
helpful, but GC tuning is a dark art :-)

On Thu, Nov 13, 2014 at 9:13 AM, Guozhang Wang  wrote:

> Hey Chen,
>
> As Neha suggested, typical reason of too many rebalances is that your
> consumers kept being timed out from ZK, and you can verify this by checking
> in your consumer logs for sth. like "session timeout" entries (these are
> not ERROR entries).
>
> Guozhang
>
> Guozhang
>
> On Wed, Nov 12, 2014 at 5:31 PM, Neha Narkhede 
> wrote:
>
> > Does this help?
> >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Whyaretheremanyrebalancesinmyconsumerlog
> > ?
> >
> > On Wed, Nov 12, 2014 at 3:53 PM, Chen Wang 
> > wrote:
> >
> > > Hi there,
> > > My kafka client is reading a 3 partition topic from kafka with 3
> threads
> > > distributed on different machines. I am seeing frequent owner changes
> on
> > > the topics when running:
> > > bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group
> > > my_test_group --topic mytopic -zkconnect localhost:2181
> > >
> > > The owner kept changing once a while, but I didn't see any exceptions
> > > thrown from the consumer side. When checking broker log, its full of
> > >  INFO Closing socket connection to /IP. (kafka.network.Processor)
> > >
> > > Is this expected behavior? If so,  how can I tell when  the leader is
> > > imbalanced, and rebalance is triggered?
> > > Thanks,
> > > Chen
> > >
> >
>
>
>
> --
> -- Guozhang
>

Re: Broker keeps rebalancing

[Neha Narkhede]

Does this help?
https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Whyaretheremanyrebalancesinmyconsumerlog
?

On Wed, Nov 12, 2014 at 3:53 PM, Chen Wang 
wrote:

> Hi there,
> My kafka client is reading a 3 partition topic from kafka with 3 threads
> distributed on different machines. I am seeing frequent owner changes on
> the topics when running:
> bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group
> my_test_group --topic mytopic -zkconnect localhost:2181
>
> The owner kept changing once a while, but I didn't see any exceptions
> thrown from the consumer side. When checking broker log, its full of
>  INFO Closing socket connection to /IP. (kafka.network.Processor)
>
> Is this expected behavior? If so,  how can I tell when  the leader is
> imbalanced, and rebalance is triggered?
> Thanks,
> Chen
>

Re: Add partitions with replica assignment in same command

[Neha Narkhede]

When we first add partitions, will it change the
assignment of replicas for existing partitions?

Nope. It should not touch the existing partitions.

Also, will there be any issues executing the second reassignment
command which will change the assignment again for the new partitions added?

No. 2nd reassignment should work as expected.

On Wed, Nov 12, 2014 at 2:24 PM, Allen Wang 
wrote:

> I found this JIRA
>
> https://issues.apache.org/jira/browse/KAFKA-1656
>
> Now, we have to use two commands to accomplish the goal - first add
> partitions using TopicCommand and then reassign replicas using
> ReassignPartitionsCommand. When we first add partitions, will it change the
> assignment of replicas for existing partitions? This is what we would like
> to avoid. Also, will there be any issues executing the second reassignment
> command which will change the assignment again for the new partitions
> added?
>
>
>
>
> On Sun, Nov 9, 2014 at 9:01 PM, Jun Rao  wrote:
>
> > Yes, it seems that we need to fix the tool to support that. It's probably
> > more intuitive to have TopicCommand just take the replica-assignment (for
> > the new partitions) when altering a topic. Could you file a jira?
> >
> > Thanks,
> >
> > Jun
> >
> > On Fri, Nov 7, 2014 at 4:17 PM, Allen Wang 
> > wrote:
> >
> > > I am trying to figure out how to add partitions and assign replicas
> using
> > > one admin command. I tried kafka.admin.TopicCommand to increase the
> > > partition number from 9 to 12 with the following options:
> > >
> > > /apps/kafka/bin/kafka-run-class.sh kafka.admin.TopicCommand
> --zookeeper
> > > ${ZOOKEEPER} --alter --topic test_topic_4 --partitions 12
> > > --replica-assignment 2:1,0:2,1:0,1:2,2:0,0:1,1:0,2:1,0:2,2:1,0:2,1:0
> > >
> > > This gives me an error
> > >
> > > Option "[replica-assignment]" can't be used with option"[partitions]"
> > >
> > > Looking into the TopicCommand, alterTopic function seems to be able to
> > > handle that but the command exits with the above error before this
> > function
> > > is invoked.
> > >
> > > Is there any workaround or other recommended way to achieve this?
> > >
> > > Thanks,
> > > Allen
> > >
> >
>

Re: expanding cluster and reassigning parititions without restarting producer

[Neha Narkhede]

The new producer is available in 0.8.2-beta (the most recent Kafka
release). The old producer only detects new partitions at an interval
configured by topic.metadata.refresh.interval.ms. This constraint is no
longer true for the new producer and you would likely end up with an even
distribution of data across all partitions. If you want to stay with the
old producer on 0.8.1.1, you can try reducing
topic.metadata.refresh.interval.ms but it may have some performance impact
on the Kafka cluster since it ends up sending topic metadata requests to
the broker at that interval.

Thanks,
Neha

On Tue, Nov 11, 2014 at 1:45 AM, Shlomi Hazan  wrote:

> Neha, I understand that the producer kafka.javaapi.producer.Producer shown
> in examples is old,
> and that a new producer (org.apache.kafka.clients.producer) is avail? is it
> available for 0.8.1.1?
> how does it work? does it have a trigger fired when partitions are added or
> does the producer refresh some cache every some given time period?
>
> Shlomi
>
>
> On Tue, Nov 11, 2014 at 4:25 AM, Neha Narkhede 
> wrote:
>
> > How can I auto refresh keyed producers to use new partitions as these
> > partitions are added?
> >
> > Try using the new producer under org.apache.kafka.clients.producer.
> >
> > Thanks,
> > Neha
> >
> > On Mon, Nov 10, 2014 at 8:52 AM, Bhavesh Mistry <
> > mistry.p.bhav...@gmail.com>
> > wrote:
> >
> > > I had different experience with expanding partition for new producer
> and
> > > its impact.  I only tried for non-key message.I would always advice
> > to
> > > keep batch size relatively low or plan for expansion with new java
> > producer
> > > in advance or since inception otherwise running producer code is
> > impacted.
> > >
> > > Here is mail chain:
> > >
> > >
> >
> http://mail-archives.apache.org/mod_mbox/kafka-dev/201411.mbox/%3ccaoejijit4cgry97dgzfjkfvaqfduv-o1x1kafefbshgirkm...@mail.gmail.com%3E
> > >
> > > Thanks,
> > >
> > > Bhavesh
> > >
> > > On Mon, Nov 10, 2014 at 5:20 AM, Shlomi Hazan 
> wrote:
> > >
> > > > Hmmm..
> > > > The Java producer example seems to ignore added partitions too...
> > > > How can I auto refresh keyed producers to use new partitions as these
> > > > partitions are added?
> > > >
> > > >
> > > > On Mon, Nov 10, 2014 at 12:33 PM, Shlomi Hazan 
> > wrote:
> > > >
> > > > > One more thing:
> > > > > I saw that the Python client is also unaffected by addition of
> > > partitions
> > > > > to a topic and that it continues to send requests only to the old
> > > > > partitions.
> > > > > is this also handled appropriately by the Java producer? Will he
> see
> > > the
> > > > > change and produce to the new partitions as well?
> > > > > Shlomi
> > > > >
> > > > > On Mon, Nov 10, 2014 at 9:34 AM, Shlomi Hazan 
> > > wrote:
> > > > >
> > > > >> No I don't see anything like that, the question was aimed at
> > learning
> > > if
> > > > >> it is worthwhile to make the effort of reimplementing the Python
> > > > producer
> > > > >> in Java, I so I will not make all the effort just to be
> disappointed
> > > > >> afterwards.
> > > > >> understand I have nothing to worry about, so I will try to
> simulate
> > > this
> > > > >> situation in small scale...
> > > > >> maybe 3 brokers, one topic with one partition and then add
> > partitions.
> > > > >> we'll see.
> > > > >> thanks for clarifying.
> > > > >> Oh, Good luck with Confluent!!
> > > > >> :)
> > > > >>
> > > > >> On Mon, Nov 10, 2014 at 4:17 AM, Neha Narkhede <
> > > neha.narkh...@gmail.com
> > > > >
> > > > >> wrote:
> > > > >>
> > > > >>> The producer might get an error code if the leader of the
> > partitions
> > > > >>> being
> > > > >>> reassigned also changes. However it should retry and succeed. Do
> > you
> > > > see
> > > > >>> a
> > > > >>> behavior that suggests otherwise?
> > > > >>>
> > > > >>> On Sat, Nov 8, 2014 at 11:45 PM, Shlomi Hazan 
> > > > wrote:
> > > > >>>
> > > > >>> > Hi All,
> > > > >>> > I recently had an issue producing from python where expanding a
> > > > cluster
> > > > >>> > from 3 to 5 nodes and reassigning partitions forced me to
> restart
> > > the
> > > > >>> > producer b/c of KeyError thrown.
> > > > >>> > Is this situation handled by the Java producer automatically or
> > > need
> > > > I
> > > > >>> do
> > > > >>> > something to have the java producer refresh itself to see the
> > > > >>> reassigned
> > > > >>> > partition layout and produce away ?
> > > > >>> > Shlomi
> > > > >>> >
> > > > >>>
> > > > >>
> > > > >>
> > > > >
> > > >
> > >
> >
>

Re: expanding cluster and reassigning parititions without restarting producer

[Neha Narkhede]

How can I auto refresh keyed producers to use new partitions as these
partitions are added?

Try using the new producer under org.apache.kafka.clients.producer.

Thanks,
Neha

On Mon, Nov 10, 2014 at 8:52 AM, Bhavesh Mistry 
wrote:

> I had different experience with expanding partition for new producer and
> its impact.  I only tried for non-key message.I would always advice to
> keep batch size relatively low or plan for expansion with new java producer
> in advance or since inception otherwise running producer code is impacted.
>
> Here is mail chain:
>
> http://mail-archives.apache.org/mod_mbox/kafka-dev/201411.mbox/%3ccaoejijit4cgry97dgzfjkfvaqfduv-o1x1kafefbshgirkm...@mail.gmail.com%3E
>
> Thanks,
>
> Bhavesh
>
> On Mon, Nov 10, 2014 at 5:20 AM, Shlomi Hazan  wrote:
>
> > Hmmm..
> > The Java producer example seems to ignore added partitions too...
> > How can I auto refresh keyed producers to use new partitions as these
> > partitions are added?
> >
> >
> > On Mon, Nov 10, 2014 at 12:33 PM, Shlomi Hazan  wrote:
> >
> > > One more thing:
> > > I saw that the Python client is also unaffected by addition of
> partitions
> > > to a topic and that it continues to send requests only to the old
> > > partitions.
> > > is this also handled appropriately by the Java producer? Will he see
> the
> > > change and produce to the new partitions as well?
> > > Shlomi
> > >
> > > On Mon, Nov 10, 2014 at 9:34 AM, Shlomi Hazan 
> wrote:
> > >
> > >> No I don't see anything like that, the question was aimed at learning
> if
> > >> it is worthwhile to make the effort of reimplementing the Python
> > producer
> > >> in Java, I so I will not make all the effort just to be disappointed
> > >> afterwards.
> > >> understand I have nothing to worry about, so I will try to simulate
> this
> > >> situation in small scale...
> > >> maybe 3 brokers, one topic with one partition and then add partitions.
> > >> we'll see.
> > >> thanks for clarifying.
> > >> Oh, Good luck with Confluent!!
> > >> :)
> > >>
> > >> On Mon, Nov 10, 2014 at 4:17 AM, Neha Narkhede <
> neha.narkh...@gmail.com
> > >
> > >> wrote:
> > >>
> > >>> The producer might get an error code if the leader of the partitions
> > >>> being
> > >>> reassigned also changes. However it should retry and succeed. Do you
> > see
> > >>> a
> > >>> behavior that suggests otherwise?
> > >>>
> > >>> On Sat, Nov 8, 2014 at 11:45 PM, Shlomi Hazan 
> > wrote:
> > >>>
> > >>> > Hi All,
> > >>> > I recently had an issue producing from python where expanding a
> > cluster
> > >>> > from 3 to 5 nodes and reassigning partitions forced me to restart
> the
> > >>> > producer b/c of KeyError thrown.
> > >>> > Is this situation handled by the Java producer automatically or
> need
> > I
> > >>> do
> > >>> > something to have the java producer refresh itself to see the
> > >>> reassigned
> > >>> > partition layout and produce away ?
> > >>> > Shlomi
> > >>> >
> > >>>
> > >>
> > >>
> > >
> >
>

Re: Rebalance not happening even after increasing max retries causing conflict in ZK

[Neha Narkhede]

A rebalance should trigger on all consumers when you add a new consumer to
the group. If you don't see the zookeeper watch fire, the consumer may have
somehow lost the watch. We have seen this behavior on older zk versions, I
wonder it that bug got reintroduced. To verify if this is the case, you can
run the wchp zookeeper command on the zk leader and check if each consumer
has a watch registered.

Do you have a way to try this on zk 3.3.4? I would recommend you try the
wchp suggestion as well.

On Fri, Nov 7, 2014 at 6:07 AM, Mohit Kathuria 
wrote:

> Hi all,
>
> Can someone help here. We are getting constant rebalance failure each time
> a consumer is added beyond a certain number. Did quite a lot of debugging
> on this and still not able to figure out the pattern.
>
> -Thanks,
> Mohit
>
> On Mon, Nov 3, 2014 at 10:53 PM, Mohit Kathuria 
> wrote:
>
> > Neha,
> >
> > Looks like an issue with the consumer rebalance not able to complete
> > successfully. We were able to reproduce the issue on topic with 30
> > partitions,  3 consumer processes(p1,p2 and p3), properties -  40
> > rebalance.max.retries and 1(10s) rebalance.backoff.ms.
> >
> > Before the process p3 was started, partition ownership was as expected:
> >
> > partitions 0-14 owned by p1
> > partitions 15-29 -> owner p2
> >
> > As the process p3 started, rebalance was triggered. Process p3 was
> > successfully able to acquire partition ownership for partitions 20-29 as
> > expected as per the rebalance algorithm. However, process p2 while trying
> > to acquire ownership of partitions 10-19 saw rebalance failure after 40
> > retries.
> >
> > Attaching the logs from process p2 and process p1. It says that p2 was
> > attempting to rebalance, it was trying to acquire ownership of partitions
> > 10-14 which were owned by process p1. However, at the same time process
> p1
> > did not get any event for giving up the partition ownership for
> partitions
> > 1-14.
> > We were expecting a rebalance to have triggered in p1 - but it didn't and
> > hence not giving up ownership. Is our assumption correct/incorrect?
> > And if the rebalance gets triggered in p1 - how to figure out apart from
> > logs as the logs on p1 did not have anything.
> >
> > *2014-11-03 06:57:36 k.c.ZookeeperConsumerConnector [INFO]
> > [topic_consumerIdString], waiting for the partition ownership to be
> > deleted: 11*
> >
> > During and after the rebalance failed on process p2, Partition Ownership
> > was as below:
> > 0-14 -> owner p1
> > 15-19 -> none
> > 20-29 -> owner p3
> >
> > This left the consumers in inconsistent state as 5 partitions were never
> > consumer from and neither was the partitions ownership balanced.
> >
> > However, there was no conflict in creating the ephemeral node which was
> > the case last time. Just to note that the ephemeral node conflict which
> we
> > were seeing earlier also appeared after rebalance failed. My hunch is
> that
> > fixing the rebalance failure will fix that issue as well.
> >
> > -Thanks,
> > Mohit
> >
> >
> >
> > On Mon, Oct 20, 2014 at 7:48 PM, Neha Narkhede 
> > wrote:
> >
> >> Mohit,
> >>
> >> I wonder if it is related to
> >> https://issues.apache.org/jira/browse/KAFKA-1585. When zookeeper
> expires
> >> a
> >> session, it doesn't delete the ephemeral nodes immediately. So if you
> end
> >> up trying to recreate ephemeral nodes quickly, it could either be in the
> >> valid latest session or from the previously expired session. If you hit
> >> this problem, then waiting would resolve it. But if not, then this may
> be
> >> a
> >> legitimate bug in ZK 3.4.6.
> >>
> >> Can you try shutting down all your consumers, waiting until session
> >> timeout
> >> and restarting them?
> >>
> >> Thanks,
> >> Neha
> >>
> >> On Mon, Oct 20, 2014 at 6:15 AM, Mohit Kathuria  >
> >> wrote:
> >>
> >> > Dear Experts,
> >> >
> >> > We recently updated to kafka v0.8.1.1 with zookeeper v3.4.5. I have of
> >> > topic with 30 partitions and 2 replicas. We are using High level
> >> consumer
> >> > api.
> >> > Each consumer process which is a storm topolofy has 5 streams which
> >> > connects to 1 or more partitions. We are not using storm's inbuilt
> kafka
> >> > spout. Everything runs fine till the 5th consumer process(25 streams)
> is
> &g

Re: Interrupting controlled shutdown breaks Kafka cluster

[Neha Narkhede]

We fixed a couple issues related to automatic leader balancing and
controlled shutdown. Would you mind trying out 0.8.2-beta?

On Fri, Nov 7, 2014 at 11:52 AM, Solon Gordon  wrote:

> We're using 0.8.1.1 with auto.leader.rebalance.enable=true.
>
> On Fri, Nov 7, 2014 at 2:35 PM, Guozhang Wang  wrote:
>
> > Solon,
> >
> > Which version of Kafka are you running and are you enabling auto leader
> > rebalance at the same time?
> >
> > Guozhang
> >
> > On Fri, Nov 7, 2014 at 8:41 AM, Solon Gordon  wrote:
> >
> > > Hi all,
> > >
> > > My team has observed that if a broker process is killed in the middle
> of
> > > the controlled shutdown procedure, the remaining brokers start spewing
> > > errors and do not properly rebalance leadership. The cluster cannot
> > recover
> > > without major manual intervention.
> > >
> > > Here is how to reproduce the problem:
> > > 1. Create a Kafka 0.8.1.1 cluster with three brokers. (Let's call them
> A,
> > > B, and C.) Set controlled.shutdown.enable=true.
> > > 2. Create a topic with replication_factor = 3 and a large number of
> > > partitions (say 100).
> > > 3. Send a TERM signal to broker A. This initiates controlled shutdown.
> > > 4. Before controlled shutdown completes, quickly send a KILL signal to
> > > broker A.
> > >
> > > Result:
> > > - Brokers B and C start logging ReplicaFetcherThread connection errors
> > > every few milliseconds. (See below for an example.)
> > > - Broker A is still listed as a leader and ISR for any partitions which
> > > were not transferred during controlled shutdown. This causes connection
> > > errors when clients try to produce to or consume from these partitions.
> > >
> > > This scenario is difficult to recover from. The only ways we have found
> > are
> > > to restart broker A multiple times (if it still exists) or to kill
> both B
> > > and C and then start them one by one. Without this kind of
> intervention,
> > > the above issues persist indefinitely.
> > >
> > > This may sound like a contrived scenario, but it's exactly what we have
> > > seen when a Kafka EC2 instance gets terminated by AWS. So this seems
> > like a
> > > real liability.
> > >
> > > Are there any existing JIRA tickets which cover this behavior? And do
> you
> > > have any recommendations for avoiding it, other than forsaking
> controlled
> > > shutdowns entirely?
> > >
> > > Thanks,
> > > Solon
> > >
> > > Error example:
> > > [2014-11-06 17:10:21,459] ERROR [ReplicaFetcherThread-0-1978259225],
> > Error
> > > in fetch Name: FetchRequest; Version: 0; CorrelationId: 3500; ClientId:
> > > ReplicaFetcherThread-0-1978259225; ReplicaId: 1359390395; MaxWait: 500
> > ms;
> > > MinBytes: 1 bytes; RequestInfo: [my-topic,42] ->
> > > PartitionFetchInfo(503,10485760),[my-topic,63] ->
> > > PartitionFetchInfo(386,10485760),[my-topic,99] ->
> > > PartitionFetchInfo(525,10485760),[my-topic,84] ->
> > > PartitionFetchInfo(436,10485760),[my-topic,48] ->
> > > PartitionFetchInfo(484,10485760),[my-topic,75] ->
> > > PartitionFetchInfo(506,10485760),[my-topic,45] ->
> > > PartitionFetchInfo(473,10485760),[my-topic,66] ->
> > > PartitionFetchInfo(532,10485760),[my-topic,30] ->
> > > PartitionFetchInfo(435,10485760),[my-topic,96] ->
> > > PartitionFetchInfo(517,10485760),[my-topic,27] ->
> > > PartitionFetchInfo(470,10485760),[my-topic,36] ->
> > > PartitionFetchInfo(472,10485760),[my-topic,9] ->
> > > PartitionFetchInfo(514,10485760),[my-topic,33] ->
> > > PartitionFetchInfo(582,10485760),[my-topic,69] ->
> > > PartitionFetchInfo(504,10485760),[my-topic,57] ->
> > > PartitionFetchInfo(444,10485760),[my-topic,78] ->
> > > PartitionFetchInfo(559,10485760),[my-topic,12] ->
> > > PartitionFetchInfo(417,10485760),[my-topic,90] ->
> > > PartitionFetchInfo(429,10485760),[my-topic,18] ->
> > > PartitionFetchInfo(497,10485760),[my-topic,0] ->
> > > PartitionFetchInfo(402,10485760),[my-topic,6] ->
> > > PartitionFetchInfo(527,10485760),[my-topic,54] ->
> > > PartitionFetchInfo(524,10485760),[my-topic,15] ->
> > > PartitionFetchInfo(448,10485760),[console,0] ->
> > > PartitionFetchInfo(4,10485760) (kafka.server.ReplicaFetcherThread)
> > > java.net.ConnectException: Connection refused
> > > at sun.nio.ch.Net.connect0(Native Method)
> > > at sun.nio.ch.Net.connect(Net.java:465)
> > > at sun.nio.ch.Net.connect(Net.java:457)
> > > at
> > sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)
> > > at
> > kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
> > > at
> kafka.consumer.SimpleConsumer.connect(SimpleConsumer.scala:44)
> > > at
> > kafka.consumer.SimpleConsumer.reconnect(SimpleConsumer.scala:57)
> > > at
> > > kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79)
> > > at
> > >
> > >
> >
> kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:71)
> > > at
> > >
> > >
> >
> kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply

Re: expanding cluster and reassigning parititions without restarting producer

[Neha Narkhede]

The producer might get an error code if the leader of the partitions being
reassigned also changes. However it should retry and succeed. Do you see a
behavior that suggests otherwise?

On Sat, Nov 8, 2014 at 11:45 PM, Shlomi Hazan  wrote:

> Hi All,
> I recently had an issue producing from python where expanding a cluster
> from 3 to 5 nodes and reassigning partitions forced me to restart the
> producer b/c of KeyError thrown.
> Is this situation handled by the Java producer automatically or need I do
> something to have the java producer refresh itself to see the reassigned
> partition layout and produce away ?
> Shlomi
>

Re: corrupt message

[Neha Narkhede]

This may be due to a bug in the client. Non-java Kafka clients are
maintained by the individual client owners. You might want to ping the
owner of your library directly.

On Mon, Nov 3, 2014 at 7:21 AM, Fredrik S Loekke 
wrote:

>  When running a C# producer against a kafka 0.8.1.1 server running on a
> virtual linux (virtualbox, Ubuntu) I keep getting the following error:
>
>
>
> [2014-11-03 15:19:08,595] ERROR [KafkaApi-0] Error processing
> ProducerRequest with correlation id 601 from client Kafka-Net on partition
> [x,0] (kafka.server.KafkaApis)
>
> kafka.message.InvalidMessageException: Message is corrupt (stored crc =
> 1767811542, computed crc = 1256103753)
>
> at kafka.message.Message.ensureValid(Message.scala:166)
>
> at
> kafka.log.Log$$anonfun$analyzeAndValidateMessageSet$1.apply(Log.scala:330)
>
> at
> kafka.log.Log$$anonfun$analyzeAndValidateMessageSet$1.apply(Log.scala:318)
>
> at scala.collection.Iterator$class.foreach(Iterator.scala:772)
>
> at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:32)
>
> at kafka.log.Log.analyzeAndValidateMessageSet(Log.scala:318)
>
> at kafka.log.Log.append(Log.scala:231)
>
> at
> kafka.cluster.Partition.appendMessagesToLeader(Partition.scala:354)
>
> at
> kafka.server.KafkaApis$$anonfun$appendToLocalLog$2.apply(KafkaApis.scala:376)
>
> at
> kafka.server.KafkaApis$$anonfun$appendToLocalLog$2.apply(KafkaApis.scala:366)
>
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
>
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
>
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
>
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
>
> at scala.collection.Iterator$class.foreach(Iterator.scala:772)
>
> at
> scala.collection.mutable.HashTable$$anon$1.foreach(HashTable.scala:157)
>
> at
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:190)
>
> at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:45)
>
> at scala.collection.mutable.HashMap.foreach(HashMap.scala:95)
>
> at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
>
> at scala.collection.mutable.HashMap.map(HashMap.scala:45)
>
> at kafka.server.KafkaApis.appendToLocalLog(KafkaApis.scala:366)
>
> at
> kafka.server.KafkaApis.handleProducerRequest(KafkaApis.scala:292)
>
> at kafka.server.KafkaApis.handle(KafkaApis.scala:185)
>
> at
> kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:42)
>
> at java.lang.Thread.run(Thread.java:745)
>
>
>
> Any suggestion on how to resolve this issue?
>
>
>
> Best regards / Med venlig hilsen
>
>
>
> *Fredrik Skeel Løkke*
>
> Software Developer ǀ IT & Analysis
>
>
>
> Mob.: +45 3176 8438
>
> f...@lindcapital.com
>
>
>
> [image: Beskrivelse: Beskrivelse: Beskrivelse: Beskrivelse:
> cid:image001.png@01CD4A0C.218B6960]
>
>
>
> Lind Capital A/S
>
> Værkmestergade 3, 2
>
> DK-8000 Aarhus C
>
> www.lindcapital.com
>
> Follow us on
>
> ­[image: linkedin]   [image:
> facebook] 
>
>
>

Re: Kafka Cluster disaster decovery

[Neha Narkhede]

A common solution for disaster recovery is to mirror the Kafka cluster into
another one deployed in a separate data center. The mirroring is not
synchronous so there might be some message loss when you lose the entire
cluster in some disaster.

Thanks,
Neha

On Mon, Nov 3, 2014 at 7:43 AM, Guozhang Wang  wrote:

> Yingkai,
>
> Kafka uses persistent storage so the data written to it will not be lost,
> you just need to restart the cluster. But during the down time it will
> become un-available.
>
> Guozhang
>
>
>
> On Fri, Oct 31, 2014 at 2:06 PM, Yingkai Hu  wrote:
>
> > Hi All,
> >
> > I’m new to Kafka, please direct me to the right path if it is a duplicate
> > question.
> >
> > Basically I deployed Kafka to a 4 machine cluster, what if the whole
> > cluster went down, does kafka provide any backup/restore mechanism?
> Please
> > advise.
> >
> > Thanks!
> > Yingkai
>
>
>
>
> --
> -- Guozhang
>

Re: Consumer lag keep increasing

[Neha Narkhede]

Chen,

Consumers lag either due to an I/O or network bottleneck or due to slow
processing of messages by the user. To confirm that you are not hitting the
latter issue, you can run a console consumer on the same data and observe
the throughput that it provides and it's lag.

Thanks,
Neha

On Wed, Nov 5, 2014 at 3:31 PM, Chen Wang 
wrote:

> Guozhang,
> I can see message keep coming, meaning messages are being consumed, right?
> But the lag is pretty huge (average 30m messages behind) as you can see
> from the graph:
>
> https://www.dropbox.com/s/xli25zicxv5f2qa/Screenshot%202014-11-05%2015.23.05.png?dl=0
>
> My understanding is that for such light weight thread, the consumer should
> almost be at the same pace with the producer. I also checked the machine
> metrics, and nothing pegged there.
>
> I am also moving the testing application to a separate dev cluster. In your
> experience, what things might cause the slow reading? Is this more like a
> server side thing, or consumer side?
>
> Chen
>
> On Wed, Nov 5, 2014 at 3:10 PM, Guozhang Wang  wrote:
>
> > Chen,
> >
> > Your configs seems fine.
> >
> > Could you use ConsumerOffsetChecker tool to see if the offset is
> advancing
> > at all (i.e. messages are comsumed), and if yes get some thread dumps and
> > check if your consumer is blocked on some locks?
> >
> > Guozhang
> >
> > On Wed, Nov 5, 2014 at 2:01 PM, Chen Wang 
> > wrote:
> >
> > > Hey Guys,
> > > I have a really simply storm topology with a kafka spout, reading from
> > > kafka through high level consumer. Since the topic has 30 partitions,
> we
> > > have 30 threads in the spout reading from it. However, it seems that
> the
> > > lag keeps increasing even the thread only read the message and do
> > nothing.
> > > The largest message size  are around 30KB, and the incoming rate can be
> > as
> > > hight as 14k/seconds. There are 3 brokers on some high config bare
> metal
> > > machines. The client side config is like this:
> > >
> > > kafka.config.fetch.message.max.bytes3145728
> > > kafka.config.group.id   spout_readonly
> > > kafka.config.rebalance.backoff.ms   6000
> > > kafka.config.rebalance.max.retries  6
> > > kafka.config.zookeeper.connect  dare-broker00.sv.walmartlabs.com:2181,
> > > dare-broker01.sv.walmartlabs.com:2181,
> > > dare-broker02.sv.walmartlabs.com:2181
> > > kafka.config.zookeeper.session.timeout.ms   6
> > >
> > > what could possibly cause this huge lag? Will broker be a bottle neck,
> or
> > > some config need to be adjusted? The server side config is like this:
> > >
> > > replica.fetch.max.bytes=2097152
> > > message.max.bytes=2097152
> > > num.network.threads=4
> > > num.io.threads=4
> > >
> > > # The send buffer (SO_SNDBUF) used by the socket server
> > > socket.send.buffer.bytes=4194304
> > >
> > > # The receive buffer (SO_RCVBUF) used by the socket server
> > > socket.receive.buffer.bytes=2097152
> > >
> > > # The maximum size of a request that the socket server will accept
> > > (protection against OOM)
> > > socket.request.max.bytes=104857600
> > >
> > > Any help appreciated!
> > > Chen
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
>

Re: Kafka Release timelines

[Neha Narkhede]

0.8.2 should be available in a month. Though 0.9 might take a couple more
months and there is a good chance that it will not be this year.

Thanks,
Neha

On Thu, Nov 6, 2014 at 3:01 AM, dinesh kumar  wrote:

> Hi,
> I found the future release plan wiki here
>  .
> I
> see that 0.8.2 is still in beta even though it was stated for September.
> What is the expected date for 0.9 release?
>
> Thanks,
> Dinesh
>

Re: Interaction of retention settings for broker and topic plus partitions

[Neha Narkhede]

To clarify though, is it correct that a per topic limit will always
override the default limit of the same type?  (e.g. a large per-topic
retention hours vs. a small default retention hours)?

That's correct.

On Thu, Nov 6, 2014 at 9:34 AM, Jason Rosenberg  wrote:

> Jun,
>
> To clarify though, is it correct that a per topic limit will always
> override the default limit of the same type?  (e.g. a large per-topic
> retention hours vs. a small default retention hours)?
>
> Jason
>
> On Sat, Sep 20, 2014 at 12:28 AM, Jun Rao  wrote:
>
> > That's right. The rule is that a log segment is deleted if either the
> size
> > or the time limit is reached. Log sizes are per partition.
> >
> > Thanks,
> >
> > Jun
> >
> > On Thu, Sep 18, 2014 at 2:55 PM, Cory Watson  wrote:
> >
> > > Hello all!
> > >
> > > I'm curious about the interaction of server and topic level retention
> > > settings. It's not clear to me the precedence of the follow:
> > >
> > >- broker's default log.retention.bytes
> > >- topic's retention.bytes (which defaults to broker's
> > >log.retention.bytes)
> > >- broker's log.retention.hours and log.retention.minutes (if both
> are
> > >specified then it seems to be the lower of the two, since it's when
> > >"either" is exceeded)
> > >
> > > It seems that the rule is that when any of these are violated then the
> > log
> > > segment is deleted. Is this right?
> > >
> > > Also, just to be clear: The log sizes in questions are for a single
> > > partitions logs?
> > >
> > > I have a situation where my per-topic retention.bytes is very high, but
> > my
> > > default log.retention.hours is lower (the default @ 168 hours). It
> seems
> > > that it's truncating at the log.retention.hours instead of the topic's
> > > retention.bytes.
> > >
> > > Am I understanding this correctly? :)
> > >
> > > --
> > > Cory Watson
> > > Principal Infrastructure Engineer // Keen IO
> > >
> >
>

Re: nulls found in topic, created by recovery?

[Neha Narkhede]

IIRC, the bug that introduced the nulls was related to compressed data. Is
this topic compressed? Did you try to run a consumer through the topic's
data or alternately the DumpLogSegments tool?

On Thu, Nov 6, 2014 at 12:56 PM, Neil Harkins  wrote:

> Hi all. I saw something weird yesterday on our "leaf" instances
> which run kafka 0.7.2 (and mirror to kafka 0.8 via our custom code).
> I fully realize everyone's instinctual response is "upgrade, already.",
> but I'd like to have an internals discussion to better understand
> what happened, as I suspect it's still relevant in 0.8.
>
> Basically, in one of our topics there was an 8k stretch of nulls.
> Correlating timestamps from the messages bracketing the nulls
> to the kafka log, I see that the server restarted during that time,
> and here are the recovery lines related to the topic with the nulls:
>
> [2014-11-04 00:48:07,602] INFO zookeeper state changed (SyncConnected)
> (org.I0Itec.zkclient.ZkClient)
> [2014-11-04 01:00:35,806] INFO Shutting down Kafka server
> (kafka.server.KafkaServer)
> [2014-11-04 01:00:35,813] INFO shutdown scheduler kafka-logcleaner-
> (kafka.utils.KafkaScheduler)
> [2014-11-04 01:01:38,411] INFO Starting Kafka server...
> (kafka.server.KafkaServer)
> ...
> [2014-11-04 01:01:49,146] INFO Loading log 'foo.bar-0'
> (kafka.log.LogManager)
> [2014-11-04 01:01:49,147] INFO Loading the last segment
> /var/kafka-leaf-spool/foo.bar-0/002684355423.kafka in mutable
> mode, recovery true (kafka.log.Log)
> [2014-11-04 01:01:55,877] INFO recover high water mark:414004449
> (kafka.message.FileMessageSet)
> [2014-11-04 01:01:55,877] INFO Recovery succeeded in 6 seconds. 0
> bytes truncated. (kafka.message.FileMessageSet)
>
> The only hypothesis I can come up with is that the shutdown
> ("graceful"?) did not wait for all messages to flush to disk
> (we're configured with: log.flush.interval=1,
> log.default.flush.interval.ms=500, and
> log.default.flush.scheduler.interval.ms=500),
> but the max offset was recorded, so that when it came back up,
> it filled the gap with nulls to reach the valid max offset in case
> any consumers were at the end.
>
> But for consumers with a position prior to all the nulls,
> are they guaranteed to get back "on the rails" so-to-speak?
> Nulls appear as v0(i.e. "magic") messages of 0 length,
> but the messages replaced could be variable length.
>
> Thanks in advance for any input,
> -neil
>

Re: High CPU usage of Crc32 on Kafka broker

[Neha Narkhede]

Allen,

Apache mailing lists don't allow attachments. Could you please link to a
pastebin or something?

Thanks,
Neha

On Thu, Nov 6, 2014 at 12:02 PM, Allen Wang 
wrote:

> After digging more into the stack trace got from flight recorder (which is
> attached), it seems that Kafka (0.8.1.1) can optimize the usage of Crc32.
> The stack trace shows that Crc32 is invoked twice from Log.append(). First
> is from the line number 231:
>
> val appendInfo = analyzeAndValidateMessageSet(messages)
>
> The second time is from line 252 in the same function:
>
> validMessages = validMessages.assignOffsets(offset, appendInfo.codec)
>
> If one of the Crc32 invocation can be eliminated, we are looking at saving
> at least 7% of CPU usage.
>
> Thanks,
> Allen
>
>
>
>
> On Wed, Nov 5, 2014 at 6:32 PM, Allen Wang  wrote:
>
>> Hi,
>>
>> Using flight recorder, we have observed high CPU usage of CRC32
>> (kafka.utils.Crc32.update()) on Kafka broker. It uses as much as 25% of CPU
>> on an instance. Tracking down stack trace, this method is invoked by
>> ReplicaFetcherThread.
>>
>> Is there any tuning we can do to reduce this?
>>
>> Also on the topic of CPU utilization, we observed that overall CPU
>> utilization is proportional to AllTopicsBytesInPerSec metric. Does this
>> metric include incoming replication traffic?
>>
>> Thanks,
>> Allen
>>
>>
>

Re: Strategies for high-concurrency consumers

[Neha Narkhede]

Jack,

Zookeeper is likely the bottleneck if rebalancing takes a very long time.
As Jay said, this will be addressed in the consumer rewrite planned for
0.9. Few more workarounds that were tried at LinkedIn - 1) To deploy
Zookeeper on SSDs and 2) Turning sync on every write off
(zookeeper.forceSync). I'm not sure if #2 negatively affected the
consistency of the zookeeper data ever but it did help with speeding up the
rebalancing.

THanks,
Neha

On Thu, Nov 6, 2014 at 11:31 AM, Jay Kreps  wrote:

> Unfortunately the performance of the consumer balancing scales poorly with
> the number of partitions. This is one of the things the consumer rewrite
> project is meant to address, however that is not complete yet. A reasonable
> workaround may be to decouple your application parallelism from the number
> of partitions. I.e. have the processing of each partition happen in a
> threadpool. I'm assuming that you don't actually have 2,500 machines, just
> that you need that much parallelism since each messages takes a bit of time
> to process. This does weaken the delivery ordering, but you may be able to
> shard the processing by key to solve that problem.
>
> -Jay
>
> On Thu, Nov 6, 2014 at 10:59 AM, Jack Foy  wrote:
>
> > Hi all,
> >
> > We are building a system that will carry a high volume of traffic (on the
> > order of 2 billion messages in each batch), which we need to process at a
> > rate of 50,000 messages per second. We need to guarantee at-least-once
> > delivery for each message. The system we are feeding has a latency of
> 50ms
> > per message, and can absorb many concurrent requests.
> >
> > We have a Kafka 0.8.1.1 cluster with three brokers and a Zookeeper 3.4.5
> > cluster with 5 nodes, each on physical hardware.
> >
> > We intend to deploy a consumer group of 2500 consumers against a single
> > topic, with a partition for each consumer. We expect our consumers to be
> > stable over the course of the run, so we expect rebalancing to be rare.
> In
> > testing, we have successfully run 512 high-level consumers against 1024
> > partitions, but beyond 512 consumers the rebalance at startup doesn’t
> > complete within 10 minutes. Is this a workable strategy with high-level
> > consumers? Can we actually deploy a consumer group with this many
> consumers
> > and partitions?
> >
> > We see throughput of more than 500,000 messages per second with our 512
> > consumers, but we need greater parallelism to meet our performance needs.
> >
> > --
> > Jack Foy 
> >
> >
> >
> >
>

Re: Dynamically adding Kafka brokers

[Neha Narkhede]

I agree that KAFKA-1070 would be great to get in. I especially felt the
need for something like this while using a few other systems that automated
the port, id etc to give a good OOTB experience. Sorry, I lost track of the
review. Will do so in the next few days.

Thanks,
Neha

On Mon, Nov 3, 2014 at 3:56 PM, Jay Kreps  wrote:

> I agree it would be really nice to get KAFKA-1070 figured out.
>
> FWIW, the reason for having a name or id other than ip was to make it
> possible to move the identity to another physical server (e.g. scp the data
> directory) and have it perform the same role on that new piece of hardware.
> Systems that tie the data to a particular host tend to be sort of hated on
> since you can't do anything simple/stupid to back them up or replace them.
>
> -Jay
>
> On Mon, Nov 3, 2014 at 2:23 PM, Gwen Shapira 
> wrote:
>
> > +1
> > Thats what we use to generate broker id in automatic deployments.
> > This method makes troubleshooting easier (you know where each broker is
> > running), and doesn't require keeping extra files around.
> >
> > On Mon, Nov 3, 2014 at 2:17 PM, Joe Stein  wrote:
> >
> > > Most folks strip the IP and use that as the broker.id. KAFKA-1070 does
> > not
> > > yet accommodate for that very widely used method. I think it would be
> bad
> > > if KAFKA-1070 only worked for new installations because that is how
> > people
> > > use Kafka today (per
> > >
> > >
> >
> https://issues.apache.org/jira/browse/KAFKA-1070?focusedCommentId=14085808&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14085808
> > > )
> > >
> > > On Mon, Nov 3, 2014 at 2:12 PM, Joel Koshy 
> wrote:
> > >
> > > > KAFKA-1070 will help with this and is pending a review.
> > > >
> > > > On Mon, Nov 03, 2014 at 05:03:20PM -0500, Otis Gospodnetic wrote:
> > > > > Hi,
> > > > >
> > > > > How do people handle situations, and specifically the broker.id
> > > > property,
> > > > > where the Kafka (broker) cluster is not fully defined right away?
> > > > >
> > > > > Here's the use case we have at Sematext:
> > > > > * Our software ships as a VM
> > > > > * All components run in this single VM, including 1 Kafka broker
> > > > > * Of course, this is just for a nice OOTB experience, but to scale
> > one
> > > > > needs to have more instances of this VM, including more Kafka
> brokers
> > > > > * *One can clone our VM and launch N instances of it, but because
> we
> > > > have a
> > > > > single Kafka broker config with a single broker.id <
> http://broker.id
> > >
> > > in
> > > > > it, one can't just launch more of these VMs and expect to see more
> > > Kafka
> > > > > brokers join the cluster.  One would have to change the broker.id
> > > > >  on each new VM instance.*
> > > > >
> > > > > How do others handle this in a software that is packages and ships
> to
> > > > user
> > > > > and is not in your direct control to allow you to edit configs?
> > > > >
> > > > > Would it be best to have a script that connect to ZooKeeper to get
> > the
> > > > list
> > > > > of all existing brokers and their IDs and then generate a new
> > distinct
> > > > ID +
> > > > > config for the new Kafka broker?
> > > > >
> > > > > Or are there slicker ways to do this that people use?
> > > > >
> > > > > Thanks,
> > > > > Otis
> > > > > --
> > > > > Monitoring * Alerting * Anomaly Detection * Centralized Log
> > Management
> > > > > Solr & Elasticsearch Support * http://sematext.com/
> > > >
> > > >
> > >
> >
>

Re: [ANNOUNCEMENT] Apache Kafka 0.8.2-beta Released

[Neha Narkhede]

Also the 0.9 consumer javadoc is here -
http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/org/apache/kafka/clients/consumer/KafkaConsumer.html

Thanks,
Neha

On Thu, Oct 30, 2014 at 5:00 AM, Joe Stein  wrote:

> Hey, yeah!
>
> For the new producer
> https://archive.apache.org/dist/kafka/0.8.2-beta/java-doc/
>
> The java consumer is slated in 0.9 more on that here
>
> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design
>
> /***
>  Joe Stein
>  Founder, Principal Consultant
>  Big Data Open Source Security LLC
>  http://www.stealth.ly
>  Twitter: @allthingshadoop 
> /
>
> On Thu, Oct 30, 2014 at 6:08 AM, Pierre-Yves Ritschard 
> wrote:
>
> > Hi Joe et al.
> >
> > Congrats on the beta release!
> > Do I read correctly that libraries can now rely on
> > org.apache.kafka/kafka-clients which does not pull in scala anymore ?
> >
> > If so, awesome!
> >
> >   - pyr
> >
> > On Tue, Oct 28, 2014 at 2:01 AM, Libo Yu  wrote:
> >
> > > Congrats! When do you think the final 0.82 will be released?
> > >
> > > > To: annou...@apache.org; users@kafka.apache.org;
> d...@kafka.apache.org
> > > > Subject: [ANNOUNCEMENT] Apache Kafka 0.8.2-beta Released
> > > > Date: Tue, 28 Oct 2014 00:50:35 +
> > > > From: joest...@apache.org
> > > >
> > > > The Apache Kafka community is pleased to announce the beta release
> for
> > > Apache Kafka 0.8.2.
> > > >
> > > > The 0.8.2-beta release introduces many new features, improvements and
> > > fixes including:
> > > >  - A new Java producer for ease of implementation and enhanced
> > > performance.
> > > >  - Delete topic support.
> > > >  - Per topic configuration of preference for consistency over
> > > availability.
> > > >  - Scala 2.11 support and dropping support for Scala 2.8.
> > > >  - LZ4 Compression.
> > > >
> > > > All of the changes in this release can be found:
> > > https://archive.apache.org/dist/kafka/0.8.2-beta/RELEASE_NOTES.html
> > > >
> > > > Apache Kafka is high-throughput, publish-subscribe messaging system
> > > rethought of as a distributed commit log.
> > > >
> > > > ** Fast => A single Kafka broker can handle hundreds of megabytes of
> > > reads and
> > > > writes per second from thousands of clients.
> > > >
> > > > ** Scalable => Kafka is designed to allow a single cluster to serve
> as
> > > the central data backbone
> > > > for a large organization. It can be elastically and transparently
> > > expanded without downtime.
> > > > Data streams are partitioned and spread over a cluster of machines to
> > > allow data streams
> > > > larger than the capability of any single machine and to allow
> clusters
> > > of co-ordinated consumers.
> > > >
> > > > ** Durable => Messages are persisted on disk and replicated within
> the
> > > cluster to prevent
> > > > data loss. Each broker can handle terabytes of messages without
> > > performance impact.
> > > >
> > > > ** Distributed by Design => Kafka has a modern cluster-centric design
> > > that offers
> > > > strong durability and fault-tolerance guarantees.
> > > >
> > > > You can download the release from:
> > > http://kafka.apache.org/downloads.html
> > > >
> > > > We welcome your help and feedback. For more information on how to
> > > > report problems, and to get involved, visit the project website at
> > > http://kafka.apache.org/
> > > >
> > >
> > >
> >
>

Re: Consumer cannot find leader after leadership changes on broker side

[Neha Narkhede]

Which Kafka version are you on for the consumer? We recently fixed a NPE
related to ZkClient, could you give 0.8.2-beta a try on your consumer and
retry the test?

On Wed, Oct 29, 2014 at 10:34 AM, Allen Wang  wrote:

> After executing PreferredReplicaLeaderElectionCommand on broker instance,
> we observed one of the consumers cannot find the leadership and stopped
> consuming. The following exception is all over the log file and it appears
> that the consumer cannot recover from it:
>
> 2014-10-29 00:53:30,492 WARN
>
> surorouter-logsummary_surorouter-logsummary-i-eaef7107-1413327811303-4afb7b23-leader-finder-thread
> ConsumerFetcherManager$LeaderFinderThread -
>
> [surorouter-logsummary_surorouter-logsummary-i-eaef7107-1413327811303-4afb7b23-leader-finder-thread],
> Failed to find leader for Set([nf_errors_log,28], [nf_errors_log,29])
> java.lang.NullPointerException
> at
> org.I0Itec.zkclient.ZkConnection.getChildren(ZkConnection.java:99)
> at org.I0Itec.zkclient.ZkClient$2.call(ZkClient.java:416)
> at org.I0Itec.zkclient.ZkClient$2.call(ZkClient.java:413)
> at
> org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
> at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:413)
> at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:409)
> at
> kafka.utils.ZkUtils$.getChildrenParentMayNotExist(ZkUtils.scala:487)
> at kafka.utils.ZkUtils$.getAllBrokersInCluster(ZkUtils.scala:84)
> at
> kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:65)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51)
>
>
> Except for this instance, other consumer instances are fine. Is there
> a workaround? Should we report it as a bug?
>
> Thanks,
> Allen
>

Re: High Level Consumer Iterator IllegalStateException Issue

[Neha Narkhede]

queued.max.message.chunks controls the consumer's fetcher queue.

On Mon, Oct 27, 2014 at 9:32 PM, Bhavesh Mistry 
wrote:

> HI Neha,
>
> If I solved the problem number 1 think and number 2 will be solved  (prob
> 1 is causing problem number 2(blocked)).  Can you please let me know what
> controls the queue size for *ConsumerFetcherThread* thread ?
>
>
> Please see the attached java source code which will reproduce the
> problem.  You may remove the recovery process...  Please check.  We have to
> do some work before we start reading from Kafka Stream Interator and this
> seems to cause some issue with java.lang.
> IllegalStateException: Iterator is in failed state*.
>
> Please let me know your finding and recommendation.
>
> Thanks,
>
> Bhavesh
>
> On Mon, Oct 27, 2014 at 6:24 PM, Neha Narkhede 
> wrote:
>
>> >> Sometime it give following exception.
>>
>> It will help to have a more specific test case that reproduces the failed
>> iterator state.
>>
>> Also, the consumer threads block if the fetcher queue is full. The queue
>> can fill up if your consumer thread dies or slows down. I'd recommend you
>> ensure that all your consumer threads are alive. You can take a thread
>> dump
>> to verify this.
>>
>> Thanks,
>> Neha
>>
>> On Mon, Oct 27, 2014 at 2:14 PM, Bhavesh Mistry <
>> mistry.p.bhav...@gmail.com>
>> wrote:
>>
>> > Hi Neha,
>> >
>> >
>> > I have two problems:.  Any help is greatly appreciated.
>> >
>> >
>> > 1)* java.lang.IllegalStateException: Iterator is in failed state*
>> >
>> >ConsumerConnector  consumerConnector = Consumer
>> > .createJavaConsumerConnector(getConsumerConfig());
>> > Map topicCountMap = new HashMap> > Integer>();
>> > topicCountMap.put(topic, *32*);
>> > Map>> topicStreamMap =
>> > consumerConnector
>> > .createMessageStreams(topicCountMap);
>> >
>> > List> streams =
>> > Collections.synchronizedList(topicStreamMap.get(topic));
>> >
>> > AppStaticInfo info = Mupd8Main.STATICINFO();
>> >
>> > Iterator> iterator =
>> > streams.iterator();
>> > // remove the head first list for this source...rest are for the
>> > Dynamic Souce...
>> > mainIterator = iterator.next().iterator();
>> >
>> > List> iteratorList = new
>> > ArrayList>(streams.size());
>> > // now rest of the iterator must be registered now..
>> > while(iterator.hasNext()){
>> > iteratorList.add(iterator.next().iterator());
>> > }
>> > *KafkaStreamRegistory.registerStream(mainSourceName,
>> > iteratorList);*
>> >
>> > Once the Consumer iterator is created and registered.  We use this in
>> > another thread to start reading from the Consumer Iterator.   Sometime
>> it
>> > give following exception.
>> >
>> > 24 Oct 2014 16:03:25,923 ERROR
>> > [SourceReader:request_source:LogStreamKafkaSource1]
>> > (grizzled.slf4j.Logger.error:116)  - SourceThread: exception during
>> reads.
>> > Swallowed to continue next read.
>> > java.lang.IllegalStateException: Iterator is in failed state
>> > at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54)
>> >
>> >
>> > I have tried to recover from this state by using this:
>> > iterator.resetState(); but it does not recover sometime.
>> >
>> >
>> >
>> >
>> > *2) ConsumerFetcherThread are blocked on enqueue ?  What controls size
>> of
>> > queue ? Why are they blocked ?  *Due to this our lags are increasing.
>> our
>> > threads blocked on hasNext()...
>> >
>> >
>> >
>> "ConsumerFetcherThread-Mupd8_Kafka_caq1-fl4-ilo.walmart.com-1414443037185-70e42954-0-1"
>> > prio=5 tid=0x7fb36292c800 nid=0xab03 waiting on condition
>> > [0x000116379000]
>> >java.lang.Thread.State: WAITING (parking)
>> > at sun.misc.Unsafe.park(Native Method)
>> > - parking to wait for  <0x000704019388> (a
>> > java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>> > at
>> > java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
>> > at
>> >
>> >
>> java.util.concurrent.locks.Abst

Re: Where Compression/Decompression happens

[Neha Narkhede]

Is Compression  happening on Producer Side (on application thread meaning
thread that call send method or background Kafka thread ) and where does
decompression Consumer side ?

Application thread in both cases.

Is there any Compression/Decompression happening on Brokers Side when
receiving message from producer and sending message to consumers  ?

Yes, before appending to the broker's log, to assign offsets to compressed
messages.

Thanks,
Neha

On Mon, Oct 27, 2014 at 9:55 PM, Bhavesh Mistry 
wrote:

> Hi Kafka Team,
>
> Is Compression  happening on Producer Side (on application thread meaning
> thread that call send method or background Kafka thread ) and where does
> decompression Consumer side ?
>
> Is there any Compression/Decompression happening on Brokers Side when
> receiving message from producer and sending message to consumers  ?
>
> Thanks for your help !
>
> Thanks,
>
> Bhavesh
>

Re: High Level Consumer Iterator IllegalStateException Issue

[Neha Narkhede]

a_caq1-fl4-ilo.walmart.com-1414443037185-70e42954-0-2"
> prio=5 tid=0x7fb36229e000 nid=0xa903 waiting on condition
> [0x000116276000]
>java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for  <0x000704064ce0> (a
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
> at
> java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
> at
>
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
> at
> java.util.concurrent.LinkedBlockingQueue.put(LinkedBlockingQueue.java:349)
> at
> kafka.consumer.PartitionTopicInfo.enqueue(PartitionTopicInfo.scala:60)
> at
>
> kafka.consumer.ConsumerFetcherThread.processPartitionData(ConsumerFetcherThread.scala:49)
> at
>
> kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$1$$anonfun$apply$mcV$sp$2.apply(AbstractFetcherThread.scala:131)
> at
>
> kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$1$$anonfun$apply$mcV$sp$2.apply(AbstractFetcherThread.scala:112)
> at
> scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:224)
> at
> scala.collection.immutable.HashMap$HashTrieMap.foreach(HashMap.scala:403)
> at
>
> kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$1.apply$mcV$sp(AbstractFetcherThread.scala:112)
> at
>
> kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$1.apply(AbstractFetcherThread.scala:112)
> at
>
> kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$1.apply(AbstractFetcherThread.scala:112)
> at kafka.utils.Utils$.inLock(Utils.scala:535)
> at
>
> kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:111)
> at
> kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:89)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
>
>
>
>
>
>
> Thanks,
>
> Bhavesh
>
>
>
> On Sun, Oct 26, 2014 at 3:14 PM, Neha Narkhede 
> wrote:
>
> > Can you provide the steps to reproduce this issue?
> >
> > On Fri, Oct 24, 2014 at 6:11 PM, Bhavesh Mistry <
> > mistry.p.bhav...@gmail.com>
> > wrote:
> >
> > > I am using one from the Kafka Trunk branch.
> > >
> > > Thanks,
> > >
> > > Bhavesh
> > >
> > > On Fri, Oct 24, 2014 at 5:24 PM, Neha Narkhede <
> neha.narkh...@gmail.com>
> > > wrote:
> > >
> > > > Which version of Kafka are you using on the consumer?
> > > >
> > > > On Fri, Oct 24, 2014 at 4:14 PM, Bhavesh Mistry <
> > > > mistry.p.bhav...@gmail.com>
> > > > wrote:
> > > >
> > > > > HI Kafka Community ,
> > > > >
> > > > > I am using kafka trunk source code and I get following exception.
> > What
> > > > > could cause the iterator to have FAILED state.  Please let me know
> > how
> > > I
> > > > > can fix this issue.
> > > > >
> > > > >
> > > > >
> > > > >
> > > > >
> > > > > *java.lang.IllegalStateException: Iterator is in failed stateat
> > > > > kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54)*
> > > > > Here is Properties:
> > > > >
> > > > > Properties props = new Properties();
> > > > > props.put("zookeeper.connect", zkConnect);
> > > > > props.put("group.id", groupId);
> > > > > *props.put("consumer.timeout.ms <
> http://consumer.timeout.ms
> > >",
> > > > > "-1");*
> > > > > props.put("zookeeper.session.timeout.ms", "1");
> > > > > props.put("zookeeper.sync.time.ms", "6000");
> > > > > props.put("auto.commit.interval.ms", "2000");
> > > > > props.put("rebalance.max.retries", "8");
> > > > > props.put("auto.offset.reset", "largest");
> > > > > props.put("fetch.message.max.bytes","2097152");
> > > > > props.put("socket.receive.buffer.bytes","2097152");
> > > > > props.put("auto.commit.enable","true");
> > > > >
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Bhavesh
> > > > >
> > > >
> > >
> >
>

Re: kafka.common.ConsumerRebalanceFailedException

[Neha Narkhede]

Can you see if the workarounds mentioned here

for failed rebalance attempts is useful?

On Fri, Oct 24, 2014 at 11:22 AM, Mohit Anchlia 
wrote:

> I am seeing following exception, don't understand the issue here. Is there
> a way to resolve this error?
>
> client consumer logs:
>
>
> Exception in thread "main" kafka.common.ConsumerRebalanceFailedException:
> groupB_ip-10-38-19-230-1414174925481-97fa3f2a can't rebalance after 4
> retries
> at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:432)
> at
>
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:722)
> at
>
> kafka.consumer.ZookeeperConsumerConnector.consume(ZookeeperConsumerConnector.scala:212)
> at kafka.javaapi.consumer.Zookeep
>
>
> server logs:
>
>
>
> [2014-10-24 14:21:47,327] INFO Got user-level KeeperException when
> processing sessionid:0x149435a553d007d type:create cxid:0x97 zxid:0xb4e
> txntype:-1 reqpath:n/a Error Path:/consumers/groupB/owners/topicA/28
> Error:KeeperErrorCode = NodeExists for /consumers/groupB/owners/topicA/28
> (org.apache.zookeeper.server.PrepRequestProcessor)
> [2014-10-24 14:21:47,329] INFO Got user-level KeeperException when
> processing sessionid:0x149435a553d007d type:create cxid:0x99 zxid:0xb4f
> txntype:-1 reqpath:n/a Error Path:/consumers/groupB/owners/topicA/23
> Error:KeeperErrorCode = NodeExists for /consumers/groupB/owners/topicA/23
> (org.apache.zookeeper.server.PrepRequestProcessor)
>

Re: High Level Consumer Iterator IllegalStateException Issue

[Neha Narkhede]

Can you provide the steps to reproduce this issue?

On Fri, Oct 24, 2014 at 6:11 PM, Bhavesh Mistry 
wrote:

> I am using one from the Kafka Trunk branch.
>
> Thanks,
>
> Bhavesh
>
> On Fri, Oct 24, 2014 at 5:24 PM, Neha Narkhede 
> wrote:
>
> > Which version of Kafka are you using on the consumer?
> >
> > On Fri, Oct 24, 2014 at 4:14 PM, Bhavesh Mistry <
> > mistry.p.bhav...@gmail.com>
> > wrote:
> >
> > > HI Kafka Community ,
> > >
> > > I am using kafka trunk source code and I get following exception.  What
> > > could cause the iterator to have FAILED state.  Please let me know how
> I
> > > can fix this issue.
> > >
> > >
> > >
> > >
> > >
> > > *java.lang.IllegalStateException: Iterator is in failed stateat
> > > kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54)*
> > > Here is Properties:
> > >
> > > Properties props = new Properties();
> > > props.put("zookeeper.connect", zkConnect);
> > > props.put("group.id", groupId);
> > > *props.put("consumer.timeout.ms <http://consumer.timeout.ms>",
> > > "-1");*
> > > props.put("zookeeper.session.timeout.ms", "1");
> > > props.put("zookeeper.sync.time.ms", "6000");
> > > props.put("auto.commit.interval.ms", "2000");
> > > props.put("rebalance.max.retries", "8");
> > > props.put("auto.offset.reset", "largest");
> > > props.put("fetch.message.max.bytes","2097152");
> > > props.put("socket.receive.buffer.bytes","2097152");
> > > props.put("auto.commit.enable","true");
> > >
> > >
> > > Thanks,
> > >
> > > Bhavesh
> > >
> >
>

Re: Example of using simple consumer to fetch from multiple partitions of a topic

[Neha Narkhede]

Is there a better way to do this where I can build one big requests with
broker -> partition mappings and call a consumer.fetch() with one giant
request?

Unfortunately, not until the 0.9 consumer is released. Until then, you can
take a look at ZookeeperConsumerConnector on how it manages multiple
SimpleConsumers.

Thanks,
Neha

On Fri, Oct 24, 2014 at 1:46 PM, Rajiv Kurian  wrote:

> I saw the example at
>
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> and it answered most of my questions. I am still trying to figure out the
> pattern to be used when I want to use a single simple consumer to fetch
> messages from different partitions (possibly from different offsets) and
> possibly managed by different leaders.
>
> My use case right now is that I have a consumer which dynamically decides
> which partitions it is responsible for and decides to fetch messages from
> them at potentially different offsets.
>
> Right now it seems like I would need a new SimpleConsumer for each broker
> since the SimpleConsumer takes the leadBroker in it's constructor. Then I'd
> have to build a FetchRequest for each broker and ensure that the
> addFetch(...) calls are made with partitions that correspond to the leader
> broker that a SimpleConsumer is managing. Finally I'll need to make up to
> numBrokers number of consumer.fetch(req) calls since each request is for a
> separate broker.
>
> Is there a better way to do this where I can build one big requests with
> broker -> partition mappings and call a consumer.fetch() with one giant
> request? Otherwise if I am doing this in a single thread I have head of
> line blocking with one request blocking another.
>
> Thanks!
>

Re: High Level Consumer Iterator IllegalStateException Issue

[Neha Narkhede]

Which version of Kafka are you using on the consumer?

On Fri, Oct 24, 2014 at 4:14 PM, Bhavesh Mistry 
wrote:

> HI Kafka Community ,
>
> I am using kafka trunk source code and I get following exception.  What
> could cause the iterator to have FAILED state.  Please let me know how I
> can fix this issue.
>
>
>
>
>
> *java.lang.IllegalStateException: Iterator is in failed stateat
> kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54)*
> Here is Properties:
>
> Properties props = new Properties();
> props.put("zookeeper.connect", zkConnect);
> props.put("group.id", groupId);
> *props.put("consumer.timeout.ms ",
> "-1");*
> props.put("zookeeper.session.timeout.ms", "1");
> props.put("zookeeper.sync.time.ms", "6000");
> props.put("auto.commit.interval.ms", "2000");
> props.put("rebalance.max.retries", "8");
> props.put("auto.offset.reset", "largest");
> props.put("fetch.message.max.bytes","2097152");
> props.put("socket.receive.buffer.bytes","2097152");
> props.put("auto.commit.enable","true");
>
>
> Thanks,
>
> Bhavesh
>

Re: Kafka 0.9

[Neha Narkhede]

The new consumer development hasn't started yet. But we have a very
detailed design doc

and JIRA  plan, if you'd
like to contribute. The new client will be protocol compatible with the 0.8
broker.

On Thu, Oct 23, 2014 at 3:54 PM, Rajiv Kurian  wrote:

> I really like the Kafka 0.9 consumer api. I want to start using it. Is it
> available on maven or maybe as a downloadable jar ? If not what is the best
> way to get it?
>
> Also wanted to ask if it the new client protocol is compatible with the 0.8
> broker.
>
> Thanks!
>

Re: Reusable consumer across consumer groups

[Neha Narkhede]

I'm wondering how much of this can be done using careful system design vs
building it within the consumer itself. You could distribute the several
consumer instances across machines since it is built for distributed load
balancing. That will sufficiently isolate the resources required to run the
various consumers. But probably you have a specific use case in mind for
running several consumer groups on the same machine. Would you mind giving
more details?

On Thu, Oct 23, 2014 at 12:55 AM, Stevo Slavić  wrote:

> Imagine exposing Kafka over various remoting protocols, where incoming
> poll/read requests may come in concurrently for different consumer groups,
> especially in a case with lots of different consumer groups.
> If you create and destroy KafkaConsumer for each such request, response
> times and throughput will be very low, and doing that is one of the ways to
> reproduce https://issues.apache.org/jira/browse/KAFKA-1716
>
> It would be better if one could reuse a (pool of) Consumer instances, and
> through a read operation parameter specify for which consumer group should
> read be performed.
>
> Kind regards,
> Stevo Slavic.
>
> On Tue, Oct 14, 2014 at 6:17 PM, Neha Narkhede 
> wrote:
>
> > Stevo,
> >
> > The new consumer API is planned for 0.9, not 0.8.2. You can take a look
> at
> > a detailed javadoc here
> > <
> >
> http://people.apache.org/~nehanarkhede/kafka-0.9-consumer-javadoc/doc/org/apache/kafka/clients/consumer/KafkaConsumer.html
> > >
> > .
> >
> > Can you explain why you would like to poll messages across consumer
> groups
> > using just one instance?
> >
> > Thanks,
> > Neha
> >
> > On Tue, Oct 14, 2014 at 1:03 AM, Stevo Slavić  wrote:
> >
> > > Hello Apache Kafka community,
> > >
> > > Current (Kafka 0.8.1.1) high-level API's KafkaConsumer is not
> lightweight
> > > object, it's creation takes some time and resources, and it does not
> seem
> > > to be thread-safe. It's API also does not support reuse, for consuming
> > > messages from different consumer groups.
> > >
> > > I see even in the coming (0.8.2) redesigned API it will not be possible
> > to
> > > reuse consumer instance to poll messages from different consumer
> groups.
> > >
> > > Can something be done to support this?
> > >
> > > Would it help if there was consumer group as a separate entity from
> > > consumer, for all the subscription management tasks?
> > >
> > > Kind regards,
> > > Stevo Slavic
> > >
> >
>

Re: How many partition can one single machine handle in Kafka?

[Neha Narkhede]

In my experience, RAID 10 doesn't really provide value in the presence of
replication. When a disk fails, the RAID resync process is so I/O intensive
that it renders the broker useless until it completes. When this happens,
you actually have to take the broker out of rotation and move the leaders
off of it to prevent it from serving requests in a degraded state. You
might as well shutdown the broker, delete the broker's data and let it
catch up from the leader.

On Wed, Oct 22, 2014 at 11:20 AM, Gwen Shapira 
wrote:

> Makes sense. Thanks :)
>
> On Wed, Oct 22, 2014 at 11:10 AM, Jonathan Weeks
>  wrote:
> > There are various costs when a broker fails, including broker leader
> election for each partition, etc., as well as exposing possible issues for
> in-flight messages, and client rebalancing etc.
> >
> > So even though replication provides partition redundancy, RAID 10 on
> each broker is usually a good tradeoff to prevent the typical most common
> cause of broker server failure (e.g. disk failure) as well, and overall
> smoother operation.
> >
> > Best Regards,
> >
> > -Jonathan
> >
> >
> > On Oct 22, 2014, at 11:01 AM, Gwen Shapira 
> wrote:
> >
> >> RAID-10?
> >> Interesting choice for a system where the data is already replicated
> >> between nodes. Is it to avoid the cost of large replication over the
> >> network? how large are these disks?
> >>
> >> On Wed, Oct 22, 2014 at 10:00 AM, Todd Palino 
> wrote:
> >>> In fact there are many more than 4000 open files. Many of our brokers
> run
> >>> with 28,000+ open files (regular file handles, not network
> connections). In
> >>> our case, we're beefing up the disk performance as much as we can by
> >>> running in a RAID-10 configuration with 14 disks.
> >>>
> >>> -Todd
> >>>
> >>> On Tue, Oct 21, 2014 at 7:58 PM, Xiaobin She 
> wrote:
> >>>
>  Todd,
> 
>  Actually I'm wondering how kafka handle so much partition, with one
>  partition there is at least one file on disk, and with 4000 partition,
>  there will be at least 4000 files.
> 
>  When all these partitions have write request, how did Kafka make the
> write
>  operation on the disk to be sequential (which is emphasized in the
> design
>  document of Kafka) and make sure the disk access is effective?
> 
>  Thank you for your reply.
> 
>  xiaobinshe
> 
> 
> 
>  2014-10-22 5:10 GMT+08:00 Todd Palino :
> 
> > As far as the number of partitions a single broker can handle, we've
> set
> > our cap at 4000 partitions (including replicas). Above that we've
> seen
>  some
> > performance and stability issues.
> >
> > -Todd
> >
> > On Tue, Oct 21, 2014 at 12:15 AM, Xiaobin She 
> > wrote:
> >
> >> hello, everyone
> >>
> >> I'm new to kafka, I'm wondering what's the max num of partition can
> one
> >> siggle machine handle in Kafka?
> >>
> >> Is there an sugeest num?
> >>
> >> Thanks.
> >>
> >> xiaobinshe
> >>
> >
> 
> >
>

Re: Performance issues

[Neha Narkhede]

the server.properties file doesn't have all the properties. You can add it
there and try your test.

On Wed, Oct 22, 2014 at 11:41 AM, Mohit Anchlia 
wrote:

> I can't find this property in server.properties file. Is that the right
> place to set this parameter?
> On Tue, Oct 21, 2014 at 6:27 PM, Jun Rao  wrote:
>
> > Could you also set replica.fetch.wait.max.ms in the broker to sth much
> > smaller?
> >
> > Thanks,
> >
> > Jun
> >
> > On Tue, Oct 21, 2014 at 2:15 PM, Mohit Anchlia 
> > wrote:
> >
> > > I set the property to 1 in the consumer code that is passed to
> > > "createJavaConsumerConnector"
> > > code, but it didn't seem to help
> > >
> > > props.put("fetch.wait.max.ms", fetchMaxWait);
> > >
> > > On Tue, Oct 21, 2014 at 1:21 PM, Guozhang Wang 
> > wrote:
> > >
> > > > This is a consumer config:
> > > >
> > > > fetch.wait.max.ms
> > > >
> > > > On Tue, Oct 21, 2014 at 11:39 AM, Mohit Anchlia <
> > mohitanch...@gmail.com>
> > > > wrote:
> > > >
> > > > > Is this a parameter I need to set it in kafka server or on the
> client
> > > > side?
> > > > > Also, can you help point out which one exactly is consumer max wait
> > > time
> > > > > from this list?
> > > > >
> > > > > https://kafka.apache.org/08/configuration.html
> > > > >
> > > > > On Tue, Oct 21, 2014 at 11:35 AM, Jay Kreps 
> > > wrote:
> > > > >
> > > > > > There was a bug that could lead to the fetch request from the
> > > consumer
> > > > > > hitting it's timeout instead of being immediately triggered by
> the
> > > > > produce
> > > > > > request. To see if you are effected by that set you consumer max
> > wait
> > > > > time
> > > > > > to 1 ms and see if the latency drops to 1 ms (or, alternately,
> try
> > > with
> > > > > > trunk and see if that fixes the problem).
> > > > > >
> > > > > > The reason I suspect this problem is because the default timeout
> in
> > > the
> > > > > > java consumer is 100ms.
> > > > > >
> > > > > > -Jay
> > > > > >
> > > > > > On Tue, Oct 21, 2014 at 11:06 AM, Mohit Anchlia <
> > > > mohitanch...@gmail.com>
> > > > > > wrote:
> > > > > >
> > > > > > > This is the version I am using: kafka_2.10-0.8.1.1
> > > > > > >
> > > > > > > I think this is fairly recent version
> > > > > > > On Tue, Oct 21, 2014 at 10:57 AM, Jay Kreps <
> jay.kr...@gmail.com
> > >
> > > > > wrote:
> > > > > > >
> > > > > > > > What version of Kafka is this? Can you try the same test
> > against
> > > > > trunk?
> > > > > > > We
> > > > > > > > fixed a couple of latency related bugs which may be the
> cause.
> > > > > > > >
> > > > > > > > -Jay
> > > > > > > >
> > > > > > > > On Tue, Oct 21, 2014 at 10:50 AM, Mohit Anchlia <
> > > > > > mohitanch...@gmail.com>
> > > > > > > > wrote:
> > > > > > > >
> > > > > > > > > It's consistently close to 100ms which makes me believe
> that
> > > > there
> > > > > > are
> > > > > > > > some
> > > > > > > > > settings that I might have to tweak, however, I am not sure
> > how
> > > > to
> > > > > > > > confirm
> > > > > > > > > that assumption :)
> > > > > > > > > On Tue, Oct 21, 2014 at 8:53 AM, Mohit Anchlia <
> > > > > > mohitanch...@gmail.com
> > > > > > > >
> > > > > > > > > wrote:
> > > > > > > > >
> > > > > > > > > > I have a java test that produces messages and then
> consumer
> > > > > > consumers
> > > > > > >

Re: Errors after reboot on single node setup

[Neha Narkhede]

Can you provide steps to reproduce this? I'm not sure I understand how you
run into this. It does look like a bug.

On Wed, Oct 22, 2014 at 9:55 AM, Ciprian Hacman  wrote:

> Hi,
>
> First of all, I am new to Kafka and more of a user than a developer. I will
> try to clarify things as much as possible though.
>
> We are using Kafka as a message system for our apps and works nicely in our
> SaaS cluster.
> I am trying to make the apps also work on a single node for demo purposes.
> I set up Zookeeper, Kafka and our apps on a node and things were ok until
> rebooting the node. After that I see the following messages in Kafka log:
>
> [2014-10-22 16:37:22,206] INFO [Controller 0]: Controller starting up
> (kafka.controller.KafkaController)
> [2014-10-22 16:37:22,419] INFO [Controller 0]: Controller startup complete
> (kafka.controller.KafkaController)
> [2014-10-22 16:37:22,554] INFO conflict in /brokers/ids/0 data:
>
> {"jmx_port":-1,"timestamp":"1413995842465","host":"ip-10-91-142-54.eu-west-1.compute.internal","version":1,"port":9092}
> stored data:
>
> {"jmx_port":-1,"timestamp":"1413994171579","host":"ip-10-91-142-54.eu-west-1.compute.internal","version":1,"port":9092}
> (kafka.utils.ZkUtils$)
> [2014-10-22 16:37:22,736] INFO I wrote this conflicted ephemeral node
>
> [{"jmx_port":-1,"timestamp":"1413995842465","host":"ip-10-91-142-54.eu-west-1.compute.internal","version":1,"port":9092}]
> at /brokers/ids/0 a while back in a different session, hence I will backoff
> for this node to be deleted by Zookeeper and retry (kafka.utils.ZkUtils$)
> [2014-10-22 16:37:25,010] ERROR Error handling event ZkEvent[Data of
> /controller changed sent to
> kafka.server.ZookeeperLeaderElector$LeaderChangeListener@a6af882]
> (org.I0Itec.zkclient.ZkEventThread)
> java.lang.IllegalStateException: Kafka scheduler has not been started
> at
> kafka.utils.KafkaScheduler.ensureStarted(KafkaScheduler.scala:114)
> at kafka.utils.KafkaScheduler.shutdown(KafkaScheduler.scala:86)
> at
>
> kafka.controller.KafkaController.onControllerResignation(KafkaController.scala:350)
> at
>
> kafka.controller.KafkaController$$anonfun$2.apply$mcV$sp(KafkaController.scala:162)
> at
>
> kafka.server.ZookeeperLeaderElector$LeaderChangeListener$$anonfun$handleDataDeleted$1.apply$mcZ$sp(ZookeeperLeaderElector.scala:138)
> at
>
> kafka.server.ZookeeperLeaderElector$LeaderChangeListener$$anonfun$handleDataDeleted$1.apply(ZookeeperLeaderElector.scala:134)
> at
>
> kafka.server.ZookeeperLeaderElector$LeaderChangeListener$$anonfun$handleDataDeleted$1.apply(ZookeeperLeaderElector.scala:134)
> at kafka.utils.Utils$.inLock(Utils.scala:535)
> at
>
> kafka.server.ZookeeperLeaderElector$LeaderChangeListener.handleDataDeleted(ZookeeperLeaderElector.scala:134)
> at org.I0Itec.zkclient.ZkClient$6.run(ZkClient.java:549)
> at org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:71)
> [2014-10-22 16:37:28,757] INFO Registered broker 0 at path /brokers/ids/0
> with address ip-10-91-142-54.eu-west-1.compute.internal:9092.
> (kafka.utils.ZkUtils$)
> [2014-10-22 16:37:28,849] INFO [Kafka Server 0], started
> (kafka.server.KafkaServer)
> [2014-10-22 16:38:56,718] INFO Closing socket connection to /127.0.0.1.
> (kafka.network.Processor)
> [2014-10-22 16:38:56,850] INFO Closing socket connection to /127.0.0.1.
> (kafka.network.Processor)
> [2014-10-22 16:38:56,985] INFO Closing socket connection to /127.0.0.1.
> (kafka.network.Processor)
>
>
> The last log line repeats forever and is correlated with errors on the app
> side.
> Restarting Kafka fixes the errors.
>
> I am using Kafka 0.8.2 from github to avoid
> https://issues.apache.org/jira/browse/KAFKA-1451.
>
> Does anyone have any idea why this happens and how it can be fixed?
>
> Thanks,
> Ciprian
> --
> Performance Monitoring * Log Analytics * Search Analytics
> Solr & Elasticsearch Support * http://sematext.com/
>

Re: frequent periods of ~1500 replicas not in sync

[Neha Narkhede]

Neil,

We fixed a bug related to the BadVersion problem in 0.8.1.1. Would you mind
repeating your test on 0.8.1.1 and if you can still reproduce this issue,
then send around the thread dump and attach the logs to KAFKA-1407?

Thanks,
Neha

On Tue, Oct 21, 2014 at 11:56 AM, Neil Harkins  wrote:

> Hi. I've got a 5 node cluster running Kafka 0.8.1,
> with 4697 partitions (2 replicas each) across 564 topics.
> I'm sending it about 1% of our total messaging load now,
> and several times a day there is a period where 1~1500
> partitions have one replica not in sync. Is this normal?
> If a consumer is reading from a replica that gets deemed
> "not in sync", does it get redirected to the good replica?
> Is there a #partitions over which maintenance tasks
> become infeasible?
>
> Relevant config bits:
> auto.leader.rebalance.enable=true
> leader.imbalance.per.broker.percentage=20
> leader.imbalance.check.interval.seconds=30
> replica.lag.time.max.ms=1
> replica.lag.max.messages=4000
> num.replica.fetchers=4
> replica.fetch.max.bytes=10485760
>
> Not necessarily correlated to those periods,
> I see a lot of these errors in the logs:
>
> [2014-10-20 21:23:26,999] 21963614 [ReplicaFetcherThread-3-1] ERROR
> kafka.server.ReplicaFetcherThread  - [ReplicaFetcherThread-3-1], Error
> in fetch Name: FetchRequest; Version: 0; CorrelationId: 77423;
> ClientId: ReplicaFetcherThread-3-1; ReplicaId: 2; MaxWait: 500 ms;
> MinBytes: 1 bytes; RequestInfo: ...
>
> And a few of these:
>
> [2014-10-20 21:23:39,555] 3467527 [kafka-scheduler-2] ERROR
> kafka.utils.ZkUtils$  - Conditional update of path
> /brokers/topics/foo.bar/partitions/3/state with data
> {"controller_epoch":11,"leader":3,"version":1,"leader_epoch":109,"isr":[3]}
> and expected version 197 failed due to
> org.apache.zookeeper.KeeperException$BadVersionException:
> KeeperErrorCode = BadVersion for
> /brokers/topics/foo.bar/partitions/3/state
>
> And this one I assume is a client closing the connection non-gracefully,
> thus should probably be a warning, not an error?:
>
> [2014-10-20 21:54:15,599] 23812214 [kafka-processor-9092-3] ERROR
> kafka.network.Processor  - Closing socket for /10.31.0.224 because of
> error
>
> -neil
>

Re: 0.8.1.2

[Neha Narkhede]

Yes, 0.8.2 includes the new producer. 0.8.2 will have a lot of new features
which will take time to stabilize. If people want 0.8.1.2 for some critical
bug fixes, we can discuss the feasibility of doing the release.

On Wed, Oct 22, 2014 at 1:39 AM, Shlomi Hazan  wrote:

> at the time I thought it was a good idea but if I understand correctly what
> Jun is saying is that 0.8.1.2 will not happen.
> I assume Jun sees 0.8.2 coming soon enough to remove any added value from
> 0.8.1.2.
> Shlomi
>
> On Wed, Oct 22, 2014 at 5:21 AM, Neha Narkhede 
> wrote:
>
> > Shlomi,
> >
> > As Jun mentioned, we are voting on a 0.8.2 beta release now. Are you
> > suggesting there be an 0.8.1.2 release in addition to that? We can take a
> > quick vote from the community to see how many people prefer to have this
> > and why.
> >
> > Thanks,
> > Neha
> >
> > On Tue, Oct 21, 2014 at 6:03 PM, Jun Rao  wrote:
> >
> > > We are voting an 0.8.2 beta release right now.
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > > On Tue, Oct 21, 2014 at 11:17 AM, Shlomi Hazan 
> wrote:
> > >
> > > > Hi All,
> > > > Will version 0.8.1.2 happen?
> > > > Shlomi
> > > >
> > >
> >
>

Re: 0.8.1.2

[Neha Narkhede]

Shlomi,

As Jun mentioned, we are voting on a 0.8.2 beta release now. Are you
suggesting there be an 0.8.1.2 release in addition to that? We can take a
quick vote from the community to see how many people prefer to have this
and why.

Thanks,
Neha

On Tue, Oct 21, 2014 at 6:03 PM, Jun Rao  wrote:

> We are voting an 0.8.2 beta release right now.
>
> Thanks,
>
> Jun
>
> On Tue, Oct 21, 2014 at 11:17 AM, Shlomi Hazan  wrote:
>
> > Hi All,
> > Will version 0.8.1.2 happen?
> > Shlomi
> >
>

Re: Performance issues

[Neha Narkhede]

Can you give more information about the performance test? Which test? Which
queue? How did you measure the dequeue latency.

On Mon, Oct 20, 2014 at 5:09 PM, Mohit Anchlia 
wrote:

> I am running a performance test and from what I am seeing is that messages
> are taking about 100ms to pop from the queue itself and hence making the
> test slow. I am looking for pointers of how I can troubleshoot this issue.
>
> There seems to be plenty of CPU and IO available. I am running 22 producers
> and 22 consumers in the same group.
>

Re: Sending Same Message to Two Topics on Same Broker Cluster

[Neha Narkhede]

I'm not sure I understood your concern about invoking send() twice, once
with each topic. Are you worried about the network overhead? Whether Kafka
does this transparently or not, sending messages to different topics will
carry some overhead. I think the design of the API is much more intuitive
and cleaner if a message is sent to a topic partition.

On Mon, Oct 20, 2014 at 9:17 PM, Bhavesh Mistry 
wrote:

> Hi Neha,
>
> Yes, I understand that but when transmitting single message (I can not set
> List of all topics)  Only Single one.  So I will to add same message in
> buffer with different topic. If Kakfa protocol, allows to add multiple
> topic then message does not have to be re-transmited over the wire to add
> to multiple topic.
>
> The Producer record only allow one topic.
>
> http://people.apache.org/~nehanarkhede/kafka-0.9-producer-javadoc/doc/org/apache/kafka/clients/producer/ProducerRecord.html
>
> Thanks for your quick response and I appreciate your help.
>
> Thanks,
>
> Bhavesh
>
>
> On Mon, Oct 20, 2014 at 9:10 PM, Neha Narkhede 
> wrote:
>
> > Not really. You need producers to send data to Kafka.
> >
> > On Mon, Oct 20, 2014 at 9:05 PM, Bhavesh Mistry <
> > mistry.p.bhav...@gmail.com>
> > wrote:
> >
> > > Hi Kakfa Team,
> > >
> > >
> > > I would like to send a single message to multiple topics (two for now)
> > > without re-transmitting the message from producer to brokers.  Is this
> > > possible?
> > >
> > > Both Producers Scala and Java does not allow this.   I do not have to
> do
> > > this all the time only based on application condition.
> > >
> > >
> > > Thanks in advance of your help !!
> > >
> > >
> > > Thanks,
> > >
> > >
> > > Bhavesh
> > >
> >
>

Re: Sending Same Message to Two Topics on Same Broker Cluster

[Neha Narkhede]

Not really. You need producers to send data to Kafka.

On Mon, Oct 20, 2014 at 9:05 PM, Bhavesh Mistry 
wrote:

> Hi Kakfa Team,
>
>
> I would like to send a single message to multiple topics (two for now)
> without re-transmitting the message from producer to brokers.  Is this
> possible?
>
> Both Producers Scala and Java does not allow this.   I do not have to do
> this all the time only based on application condition.
>
>
> Thanks in advance of your help !!
>
>
> Thanks,
>
>
> Bhavesh
>

Re: log.cleanup.interval.mins still valid for 0.8.1?

[Neha Narkhede]

Which example are you referring to?

On Mon, Oct 20, 2014 at 7:47 AM, Libo Yu  wrote:

> Hi all,
>
>
> This config property does not appear in the table of broker config
> properties. But it appears in the example on the Web page. So I wonder if
> this is still a valid config property for 0.8.1. Thanks.
>
> Libo
>
>

Re: How to produce and consume events in 2 DCs?

[Neha Narkhede]

Another way to set up this kind of mirroring is by deploying 2 clusters in
each DC - a local Kafka cluster and an aggregate Kafka cluster. The mirror
maker copies data from both the DC's local clusters into the aggregate
clusters. So if you want access to a topic with data from both DC's, you
subscribe to the aggregate cluster.

Thanks,
Neha

On Mon, Oct 20, 2014 at 7:07 AM, Erik van oosten <
e.vanoos...@grons.nl.invalid> wrote:

> Hi,
>
> We have 2 data centers that produce events. Each DC has to process events
> from both DCs.
>
> I had the following in mind:
>
>DC 1 | DC 2
> events  |events
>+  +  +  |   +  +  +
>|  |  |  |   |  |  |
>v  v  v  |   v  v  v
>  ++ | ++
>  | Receiver topic | | | Receiver topic |
>  ++   ++
>  |  |   mirroring  ||
>  |  |   +--+|
>  |  |   |   |
>  |  ++  |
>  v  vv  v
>  ++ | ++
>  | Consumer topic | | | Consumer topic |
>  ++ | ++
>+  +  +  |   +  +  +
>|  |  |  |   |  |  |
>v  v  v  |   v  v  v
>   consumers |  consumers
>
>
> As each DC has a single Kafka cluster, on each DC the receiver topic and
> consumer topic needs to be on the same cluster.
> Unfortunately, mirror maker does not seem to support mirroring to a topic
> with another name.
>
> Is there another tool we could use?
> Or, is there another approach for producing and consuming from 2 DCs?
>
> Kind regards,
> Erik.
>
> —
> Erik van Oosten
> http://www.day-to-day-stuff.blogspot.nl/
>
>

Re: taking broker down and returning it does not restore cluster state (nor rebalance)

[Neha Narkhede]

Did you ensure that your replication factor was set higher than 1? If so,
things should recover automatically after adding the killed broker back
into the cluster.

On Mon, Oct 20, 2014 at 1:32 AM, Shlomi Hazan  wrote:

> Hi,
>
> Running some tests on 0811 and wanted to see what happens when a broker is
> taken down with 'kill'. I bumped into the situation at the subject where
> launching the broker back left him a bit out of the game as far as I could
> see using stack driver metrics.
> Trying to rebalance with "verify consumer rebalance" return an error "no
> owner for partition" for all partitions of that topic (128 partitions).
> moreover, yet aside from the issue at hand, changing the group name to a
> non-existent group returned success.
> taking both the consumers and producers down allowed the rebalance to
> return success...
>
> And the question is:
> How do you restore 100% state after taking down a broker? what is the best
> practice? what needs be checked and what needs be done?
>
> Shlomi
>

Re: Rebalance not happening even after increasing max retries causing conflict in ZK

[Neha Narkhede]

Mohit,

I wonder if it is related to
https://issues.apache.org/jira/browse/KAFKA-1585. When zookeeper expires a
session, it doesn't delete the ephemeral nodes immediately. So if you end
up trying to recreate ephemeral nodes quickly, it could either be in the
valid latest session or from the previously expired session. If you hit
this problem, then waiting would resolve it. But if not, then this may be a
legitimate bug in ZK 3.4.6.

Can you try shutting down all your consumers, waiting until session timeout
and restarting them?

Thanks,
Neha

On Mon, Oct 20, 2014 at 6:15 AM, Mohit Kathuria 
wrote:

> Dear Experts,
>
> We recently updated to kafka v0.8.1.1 with zookeeper v3.4.5. I have of
> topic with 30 partitions and 2 replicas. We are using High level consumer
> api.
> Each consumer process which is a storm topolofy has 5 streams which
> connects to 1 or more partitions. We are not using storm's inbuilt kafka
> spout. Everything runs fine till the 5th consumer process(25 streams) is
> added for this topic.
>
> As soon as the sixth consumer process is added, the newly added partition
> does not get the ownership of the partitions that it requests for as the
> already existing owners have not yet given up the ownership.
>
> We changed certain properties on consumer :
>
> 1. Max Rebalance attempts - 20 ( rebalance.backoff.ms *
> rebalance.max.retries >> zk connection timeout)
> 2. Back off ms between rebalances - 1 (10seconds)
> 3. ZK connection timeout - 100,000 (100 seconds)
>
> Although when I am looking in the zookeeper shell when the rebalance is
> happening, the consumer is registered fine on the zookeeper. Just that the
> rebalance does not happen.
> After the 20th rebalance gets completed, we get
>
>
> *2014-10-11 11:10:08 k.c.ZookeeperConsumerConnector [INFO]
> [rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b], Committing all
> offsets after clearing the fetcher queues*
> *2014-10-11 11:10:10 c.s.m.k.i.c.KafkaFeedStreamer [WARN] Ignoring
> exception while trying to start streamer threads:
> rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b can't rebalance after
> 20 retries*
> *kafka.common.ConsumerRebalanceFailedException:
> rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b can't rebalance after
> 20 retries*
> *at
>
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:432)
> ~[stormjar.jar:na]*
> *at
>
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:722)
> ~[stormjar.jar:na]*
> *at
>
> kafka.consumer.ZookeeperConsumerConnector.consume(ZookeeperConsumerConnector.scala:212)
> ~[stormjar.jar:na]*
> *at
>
> kafka.javaapi.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.scala:80)
> ~[stormjar.jar:na]*
> *at
>
> com.spr.messaging.kafka.impl.consumer.KafkaFeedStreamer.createAndStartThreads(KafkaFeedStreamer.java:79)
> ~[stormjar.jar:na]*
> *at
>
> com.spr.messaging.kafka.impl.consumer.KafkaFeedStreamer.startKafkaStreamThreadsIfNecessary(KafkaFeedStreamer.java:64)
> ~[stormjar.jar:na]*
> *at
>
> com.spr.messaging.kafka.impl.consumer.KafkaFeedConsumerFactoryImpl.startStreamerIfNotRunning(KafkaFeedConsumerFactoryImpl.java:71)
> [stormjar.jar:na]*
> *at
>
> com.spr.messaging.kafka.impl.consumer.KafkaFeedPullConsumerImpl.startStreamerIfNotRunning(KafkaFeedPullConsumerImpl.java:48)
> [stormjar.jar:na]*
> *at
>
> com.spr.messaging.kafka.impl.KafkaFeedServiceImpl.getKafkaFeedPullConsumer(KafkaFeedServiceImpl.java:63)
> [stormjar.jar:na]*
> *at
>
> com.spr.storm.topology.spout.AbstractSprKafkaSpout.nextTuple(AbstractSprKafkaSpout.java:121)
> [stormjar.jar:na]*
> *at
>
> backtype.storm.daemon.executor$eval3848$fn__3849$fn__3864$fn__3893.invoke(executor.clj:562)
> [na:0.9.1-incubating]*
> *at backtype.storm.util$async_loop$fn__384.invoke(util.clj:433)
> [na:0.9.1-incubating]*
> *at clojure.lang.AFn.run(AFn.java:24) [clojure-1.4.0.jar:na]*
> *at java.lang.Thread.run(Thread.java:745) [na:1.7.0_55]*
> *2014-10-11 11:10:10 k.c.ZookeeperConsumerConnector [INFO]
> [rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b], begin registering
> consumer rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b in ZK*
> *2014-10-11 11:10:10 k.u.ZkUtils$ [INFO] conflict in
>
> /consumers/rule-engine-feed/ids/rule-engine-feed_ip-10-0-2-170-1413025767369-4679959b
> data:
>
> {"version":1,"subscription":{"rule-engine-feed":5},"pattern":"static","timestamp":"1413025810635"}
> stored data:
>
> {"version":1,"subscription":{"rule-engine-feed":5},"pattern":"static","timestamp":"1413025767483"}*
> *2014-10-11 11:10:10 k.u.ZkUtils$ [INFO] I wrote this conflicted ephemeral
> node
>
> [{"version":1,"subscription":{"rule-engine-feed":5},"pattern":"static","timestamp":"1413025810635"}]
> at
>
> /consumers/rule-engine-feed/ids/rule-e