Re: Which book to read?

[Todd Palino]

The second edition is not complete yet. The chapters that have been
released as part of the early release are updated, and you can use those
instead of the chapters in the first edition.

So use both for now :)

-Todd

On Mon, Apr 12, 2021 at 10:24 AM SuarezMiguelC
 wrote:

> Hello Apache Kafka Community!... A quick question:
>
> I am using right now kafka a LOT (with kafka connect, streams with
> nodejs...) in my arquitecture and, my knowledge has increased because of
> this, however, I'm looking for more experience, so, I downloaded the Kakfa
> book, my question is: Since there are 2 volumes, is the second just a
> better update?, or should I read the first one too?.
>
> Thanks in advance,
> Miguel Suárez

-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Capacity Engineering



linkedin.com/in/toddpalino

Re: Kafka Definitive guide v2 states auto.leader.rebalance.enable = true is not recommended

[Todd Palino]

As a note, that part of the second edition has not been updated yet. This
setting used to cause significant problems, but more recent updates to the
controller code have made the auto leader rebalancing usable.

-Todd

On Mon, Apr 12, 2021 at 10:20 AM Liam Clarke-Hutchinson <
liam.cla...@adscale.co.nz> wrote:

> Hi all,
>
> This question arose elsewhere, and I'm also going to fire it off to
> O'Reilly in the hopes that they'll clarify, but on page 180 of the
> Definitive Guide v2
> <
> https://assets.confluent.io/m/2849a76e39cda2bd/original/20201119-EB-Kafka_The_Definitive_Guide-Preview-Chapters_1_thru_6.pdf
> >
> it
> states:
>
> *Kafka brokers do not automatically take partition leadership back (unless
> auto leader rebalance is enabled, but this configuration is not
> recommended)*
>
> The original commenter raised the point that this defaults to true, and it
> sounds like a good idea to have auto leader rebalancing.
>
> So I'm curious, in anyone's war stories or experiences, has this property
> being enabled been harmful? From the context that the paragraph was written
> in, I'm assuming the writers were perhaps intending to emphasise the Cruise
> Control or Confluents self-balancing-cluster / auto-balancing features were
> preferable, but in my very brief Google didn't see any advice to set
> auto.leader.rebalance.enabled to false to use those tools.
>
> So yeah, just curious if this rings any bells.
>
> Cheers,
>
> Liam Clarke-Hutchinson
>
-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Capacity Engineering



linkedin.com/in/toddpalino

Re: On the number of partitions.

[Todd Palino]

Going to pile on here, and also say take that blog post with a grain of
salt :)

Look at your data size, and your desired degree of parallelism for
consumers, and guide to that. Yes, 1000 partitions takes a while to create.
But unless your data size is something like 1 TB per day or more, you
probably don't need that.

-Todd


On Sat, Oct 26, 2019, 1:19 PM Edward Capriolo  wrote:

> On Saturday, October 26, 2019, M. Manna  wrote:
>
> > You should also check out Becket Qin’s presentation on producer
> performance
> > tuning on YouTube. Both these items should give you all positives and
> > negatives of having many/less portions.
> >
> > Thanks,
> >
> > On Sat, 26 Oct 2019 at 09:19, Manasvi Gupta  wrote:
> >
> > >
> > > https://www.confluent.io/blog/how-choose-number-topics-
> > partitions-kafka-cluster
> > >
> > > This might be useful for you.
> > >
> > > On Sat 26 Oct, 2019, 12:40 PM Jungyeol Lee, 
> > > wrote:
> > >
> > > > I'm running a kafka cluster running only one broker with GCP
> > > n1-standard-2
> > > > instance.
> > > > I configured the number of partitions to 1000, but it is suffering
> from
> > > > running a broker on startup. It seems that it needs much more time
> for
> > > > topic creation. So, what I did is just set the smaller number of
> > > partitions
> > > > which is now 32. Everything works okay for now. Is this normal
> behavior
> > > or
> > > > performance? Or, are there any guidelines on setting the number of
> > > > partitions?
> > > >
> > > > Best,
> > > > --
> > > >
> > >
> >
>
> To many partitions is too many files and folders and file systems  and
> longer failovers.
>
>
> --
> Sorry this was sent from mobile. Will do less grammar and spell check than
> usual.
>

Re: Doubts in Kafka

[Todd Palino]

OK, in that case you’ll want to do something like use the sensor ID as the
key of the message. This will assure that every message for that sensor ID
ends up in the same partition (which will assure strict ordering of
messages for that sensor ID).

Then you can create a number of partitions to get the parallelism you
desire. For example, if you anticipate having no more than 1000 message
processors, you would create 1000 partitions. In this way, each processor
can consume messages from a single partition. In addition, you could work
up to that point. You could have 10 processors to start with, and each
would consume from 100 partitions. They would receive messages from each
partition in order (for that partition), so you will assure serial
processing of each sensor.

Note that I wouldn’t create more than 1000 partitions or so for a single
topic - it tends to give the rebalancing algorithms headaches and slow down
consumer rebalances above that. Also, you want to set up the topics with
the number of partitions once, and not expand the number of partitions
later. When you expand partitions, the affinity of key to partition
changes, so you may end up with out of order messages for a short period of
time when you expand.

-Todd

On Tue, Jan 8, 2019 at 11:11 AM aruna ramachandran 
wrote:

> I need to process single sensor messages in serial (order of messages
> should not be changed)at the same time I have to process 1 sensors
> messages in parallel please help me to configure the topics and partitions.
>
> On Tue, Jan 8, 2019 at 9:19 PM Todd Palino  wrote:
>
> > I think you’ll need to expand a little more here and explain what you
> mean
> > by processing them in parallel. Nearly by definition, parallelization and
> > strict ordering are mutually exclusive concepts.
> >
> > -Todd
> >
> > On Tue, Jan 8, 2019 at 10:40 AM aruna ramachandran <
> arunaeie...@gmail.com>
> > wrote:
> >
> > > I need to process the 1 sensor messages in parallel but each sensor
> > > message should be in order.If I create 1 partition it doesn't give
> > high
> > > throughput .Order is guaranteed only inside the partition. How can
> > > parallelize messages without changing the order pls help me to find the
> > > solution.
> > >
> >
> >
> > --
> > *Todd Palino*
> > Senior Staff Engineer, Site Reliability
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
> >
>


-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Capacity Engineering



linkedin.com/in/toddpalino

Re: Doubts in Kafka

[Todd Palino]

I think you’ll need to expand a little more here and explain what you mean
by processing them in parallel. Nearly by definition, parallelization and
strict ordering are mutually exclusive concepts.

-Todd

On Tue, Jan 8, 2019 at 10:40 AM aruna ramachandran 
wrote:

> I need to process the 1 sensor messages in parallel but each sensor
> message should be in order.If I create 1 partition it doesn't give high
> throughput .Order is guaranteed only inside the partition. How can
> parallelize messages without changing the order pls help me to find the
> solution.
>


-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Burrow V3 - going down with memory issues

[Todd Palino]

Hey, Srinivasa. It sounds like you’re running an intermediate version of
the master branch (I remember that specific error as I was making some
changes). It should be resolved with the latest version of master. Can you
try pulling the latest master?

We’ll be cutting a new release version soon, as there have been a number of
changes and updates that are useful. In the meantime, you may find more
help if you post to our Gitter:
https://gitter.im/linkedin-Burrow/Lobby

-Todd


On Thu, Mar 1, 2018 at 11:58 PM, Srinivasa Balaji <
srinivasa_bal...@trimble.com> wrote:

> We are running burrow version3 for kafka consumer lag monitoring,
>
> Issue is burrow service is getting down frequently which is every 1hr,
>
> Command used to start the burrow service
>
> # nohup $GOPATH/bin/Burrow --config-dir /opt/work/src/github.com/
> linkedin/Burrow/config 1>&2 &
>
> We are seeing the below mentioned error in log
>
> 
>
> panic: interface conversion: interface {} is nil, not *storage.brokerOffset
>
> goroutine 85 [running]:
> github.com/linkedin/Burrow/core/internal/storage.(*
> InMemoryStorage).fetchTopic(0xc4201d65a0, 0xc42100b8f0, 0xc42034d800)
> /opt/work/src/github.com/linkedin/Burrow/core/internal/
> storage/inmemory.go:611 +0x3c8
> github.com/linkedin/Burrow/core/internal/storage.(*
> InMemoryStorage).(github.com/linkedin/Burrow/core/internal/
> storage.fetchTopic)-fm(0xc42100b8f0, 0xc42034d800)
> /opt/work/src/github.com/linkedin/Burrow/core/internal/
> storage/inmemory.go:182 +0x3e
> github.com/linkedin/Burrow/core/internal/storage.(*InMemoryStorage).
> requestWorker(0xc4201d65a0, 0x11, 0xc4201d4900)
> /opt/work/src/github.com/linkedin/Burrow/core/internal/
> storage/inmemory.go:190 +0x105b
> created by github.com/linkedin/Burrow/core/internal/storage.(*
> InMemoryStorage).Start
> /opt/work/src/github.com/linkedin/Burrow/core/internal/
> storage/inmemory.go:144 +0x28b
>
>
> 
>
>
> Kindly let us know your thoughts on this to fix it.
>
> ———-
> <http://www.trimble.com/>
> Srinivasa Balaji L
> Principal Architect Cloud & DevOPS - TPaaS
> 10368, Westmoor Drive, Westminster, CO 80021
> *M*: +1(303) 324-9822 <+919790804422>
> *Email*: lsbal...@trimble.com
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Number of partition on a single broker

[Todd Palino]

Not recommended. You’ll have timeout issues with the size of the controller
requests. Additionally, there appear to be problems with writing some nodes
in Zookeeper at high partition counts.

-Todd


On Thu, Dec 14, 2017 at 8:58 AM, Abhimanyu Nagrath <
abhimanyunagr...@gmail.com> wrote:

> Can I have 20k partition on a single kafka broker ?
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

New release of Burrow!

[Todd Palino]

Hey, all. I wanted to let everyone know that I just released a brand new
version of Burrow that resolves a lot of issues that have been collecting.
We’ve been working hard on this internally for the last few months, and
it’s been running in production at LinkedIn for a few weeks now. So I’m
happy to announce that I’ve just released version 1.0.0

https://github.com/linkedin/Burrow/releases/latest

First off, from a user point of view, there’s a few big things:
1) The releases are now built for Linux, Windows, and OS X and you can
download them directly at https://github.com/linkedin/Burrow/releases/latest
2) The Docker image is now built and pushed as toddpalino/burrow
3) The config has changed significantly. Please review
https://github.com/linkedin/Burrow/wiki/Configuration

For features, there’s a lot of change:
1) Full support for topic deletion in Kafka
2) Full support for both TLS and SASL
2) Offset handling and evaluation is much cleaner and less prone to missing
commits and false alerts
3) Lag numbers are now intuitive - if a consumer stops, the lag will
increase

>From a developer point of view, this is a ground-up rewrite of Burrow:
1) Everything is modular, making it easier to add modules for custom
consumer types, or notifiers
2) The internals have a significant amount of test coverage, meaning it’s
easier to know when a change will break things
3) CI is finally set up for testing on every PR, and for building releases
4) Burrow can be started as a library. This will make it easier to build
applications that wrap it for custom configuration or logging systems.

There’s a number of things to be worked on still, and I’ll be starting
these as I have time:
1) Metrics for Burrow itself
2) Making it easier to provide custom modules without a fork
3) Better docs for setting up notifiers, and more sample configs and
templates

As always, please let me know if there are any issues. I’ve set up #burrow
on the Confluent Community Slack for discussion, though I will probably be
setting up Gitter on the project directly as well. And for bugs, there are
always the Github issues, or PRs for contributing!

-Todd


-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Change replication factor for a topic in the runtime

[Todd Palino]

You can do this using the kafka-reassign-partitions tool (or using a 3rd
party tool like kafka-assigner in github.com/linkedin/kafka-tools) to
explicitly assign the partitions to an extra replica, or remove a replica.

-Todd


On Tue, Sep 19, 2017 at 3:45 PM, Devendar Rao <devendar.gu...@gmail.com>
wrote:

> Is it possible to change the replication factor in runtime? We're using
> 10.x version.
>
> Thanks,
> Devendar
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka MirrorMaker - target or source datacenter deployment

[Todd Palino]

Always in the target datacenter. While you can set up mirror maker for no
data loss operation, it’s still a good idea to put the connection more
likely to fail (remote) on the consumer side. Additionally, there are
significant performance problems with setting it up for remote produce as
you must run with one in flight batch in order to maintain message ordering.

-Todd


On Thu, Sep 14, 2017 at 9:46 PM, Vu Nguyen <vuzilla...@gmail.com> wrote:

> Many of the descriptions and diagrams online describe deploying Kafka
> MirrorMaker into the target data center (near the target Kafka cluster).
> Since MirrorMaker is supposed to not lose messages, does it matter which
> data center MirrorMaker is deployed in--source or target data center (with
> any Kafka MirrorMaker version 0.10.1+)?
>
> It might be easier to collect and observe metrics in the source data center
> if MirrorMaker is also in the source data center (near the source Kafka
> cluster), especially if I can't aggregate the metrics from the 2 data
> centers.  Is there anything else that would influence me to deploy in
> MirrorMaker in either data center?
>
> Thanks
>
> Vu
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Avoid jvm swapping

[Todd Palino]

In production, you probably want to avoid stacking up the applications like
this. There’s a number of reasons:
1) Kafka’s performance is significantly increased by other applications not
polluting the OS page cache
2) Zookeeper has specific performance requirements - among them are a
dedicated disk for transaction logs that it can sequentially write to
3) Mirror maker chews up a lot of CPU and memory with decompression and
recompression of messages

Particular sizing of your systems is going to be dependent on the amount of
data you are moving around, but at the very least I would recommend that
your Kafka brokers, Zookeeper ensemble, and mirror makers be on separate
systems (stacking up the mirror makers on a common system is fine,
however). The Kafka brokers will need CPU and memory, and of course decent
storage to meet your retention and performance requirements. ZK needs a bit
of memory, and very good disk for the transaction logs, but it’s CPU
requirements are pretty light. Mirror maker needs CPU and memory, but it
has no real need of disk performance at all.

Sizing the brokers, you can probably get away with 3 or 4 GB of heap (this
is based on my experience running really large clusters at LinkedIn - even
at that heap size we were good for a long time), using G1 garbage
collection. The guidelines in the Kafka documentation for this are the ones
that I have developed over the last few years here. Reserve the rest of the
memory for the OS to manage - buffers and cache is your friend.

-Todd


On Mon, Aug 7, 2017 at 11:06 AM, Gabriel Machado <gmachado@gmail.com>
wrote:

> Thanks Todd, i will set swapiness to 1.
>
> Theses machines will be the future production cluster for our main
> datacenter . We have 2 remote datacenters.
> Kafka will bufferize logs and elasticsearch will index its.
>
> Is it a bad practice to have all these JVMs on the same virtual machine ?
> What do you recommend (number of machines, quantity of GB, CPU...) ? For
> the moment, each node has 4 vcpu.
>
> Gabriel.
>
> 2017-08-07 15:45 GMT+02:00 Todd Palino <tpal...@gmail.com>:
>
> > To avoid swap you should set swappiness to 1, not 0. 1 is a request
> (don't
> > swap if avoidable) whereas 0 is a demand (processes will be killed as OOM
> > instead of swapping.
> >
> > However, I'm wondering why you are running such large heaps. Most of the
> ZK
> > heap is used for storage of the data in memory, and it's obvious from
> your
> > setup that this is a development instance. So if ZK is only being used
> for
> > that Kafka cluster you are testing, you can go with a smaller heap.
> >
> > Also, for what reason are you running a 12 GB heap for Kafka? Even our
> > largest production clusters at LinkedIn are using a heap size of 6 GB
> right
> > now. You want to leave memory open for the OS to use for buffers and
> cache
> > in order to get better performance from consumers. You can see from that
> > output that it's trying to.
> >
> > It really looks like you're just overloading your system. In which case
> > swapping is to be expected.
> >
> > -Todd
> >
> >
> >
> > On Aug 7, 2017 8:34 AM, "Gabriel Machado" <gmachado@gmail.com>
> wrote:
> >
> > Hi,
> >
> > I have a 3 nodes cluster with 18 GB RAM and 2 GB swap.
> > Each node have the following JVMs (Xms=Xmx) :
> > - Zookeeper 2GB
> > - Kafka 12 GB
> > - Kafka mirror-maker DCa 1 GB
> > - Kafka mirror-maker DCb 1 GB
> >
> > All th JVMs consume 16 GB. It leaves 2 GB for the OS (debian jessie 64
> > bits).
> > Why i have no swap free on these virtual machines ?
> >
> > #free -m
> >  total   used   free sharedbuffers cached
> > Mem: 18105  17940164  0 38   
> > -/+ buffers/cache:  11235   6869
> > Swap: 2047   2045  2
> >
> >
> > I've read i should avoid jvm swapping.
> > What is the best way to do that ?
> > - modify swapiness threshold
> > - unmount all swap partition
> > - force the jvm to stay in memory with mlockall (
> > https://github.com/LucidWorks/mlockall-agent)
> > - Other solution
> >
> > Gabriel.
> >
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Avoid jvm swapping

[Todd Palino]

To avoid swap you should set swappiness to 1, not 0. 1 is a request (don't
swap if avoidable) whereas 0 is a demand (processes will be killed as OOM
instead of swapping.

However, I'm wondering why you are running such large heaps. Most of the ZK
heap is used for storage of the data in memory, and it's obvious from your
setup that this is a development instance. So if ZK is only being used for
that Kafka cluster you are testing, you can go with a smaller heap.

Also, for what reason are you running a 12 GB heap for Kafka? Even our
largest production clusters at LinkedIn are using a heap size of 6 GB right
now. You want to leave memory open for the OS to use for buffers and cache
in order to get better performance from consumers. You can see from that
output that it's trying to.

It really looks like you're just overloading your system. In which case
swapping is to be expected.

-Todd



On Aug 7, 2017 8:34 AM, "Gabriel Machado"  wrote:

Hi,

I have a 3 nodes cluster with 18 GB RAM and 2 GB swap.
Each node have the following JVMs (Xms=Xmx) :
- Zookeeper 2GB
- Kafka 12 GB
- Kafka mirror-maker DCa 1 GB
- Kafka mirror-maker DCb 1 GB

All th JVMs consume 16 GB. It leaves 2 GB for the OS (debian jessie 64
bits).
Why i have no swap free on these virtual machines ?

#free -m
 total   used   free sharedbuffers cached
Mem: 18105  17940164  0 38   
-/+ buffers/cache:  11235   6869
Swap: 2047   2045  2


I've read i should avoid jvm swapping.
What is the best way to do that ?
- modify swapiness threshold
- unmount all swap partition
- force the jvm to stay in memory with mlockall (
https://github.com/LucidWorks/mlockall-agent)
- Other solution

Gabriel.

Re: Tuning up mirror maker for high thruput

[Todd Palino]

We haven’t had any problem after tuning the default send/receive buffers in
the OS up to 10MB. Linux uses a sliding window, so if you have short
latencies, you won’t use as much of the buffer and you should see very
little, if any, impact.

-Todd


On Mon, Jul 24, 2017 at 2:20 PM, James Cheng <wushuja...@gmail.com> wrote:

> Todd,
>
> I have a question about the OS/broker tuning that you are talking about on
> the source cluster. Aside from mirrormaker (which you say should be running
> in the remote destination datacenter), presumably there will be other
> consumers in the source datacenter as well. How does the OS/broker tuning
> affect those consumers that are close to the source datacenter? Will they
> continue to function well?
>
> -James
>
> > On Jul 23, 2017, at 7:16 AM, Todd Palino <tpal...@gmail.com> wrote:
> >
> > One of the best pieces of advice I can offer is that you really need to
> run
> > the mirror maker in the same physical/network location as the Kafka
> cluster
> > you are producing to. Latency on the consumer side can be more easily
> > absorbed than latency on the producer side, as to assure that we have
> > proper message ordering and reliability, we need to restrict in flight
> > batches to 1. So that means that our produce connection is contstrained
> to
> > be very thin, and latency makes a huge impact. Meanwhile, on the consume
> > side we’re fetching large batches of messages, many at a time, so
> > round-trip latency has less of an impact. I really can’t stress this
> > enough. We set up some mirror makers in the opposite configuration for
> > security reasons, and it’s been a huge problem even with tuning.
> >
> > In addition to this, you will want to assure that your OS (and then the
> > mirror maker and broker) tuning is taking into account the latency.
> Here’s
> > a good reference for the OS side (for Linux):
> > http://linuxczar.net/blog/2016/09/18/bandwidth-delay-product/
> >
> > Once you have the OS tuned, you’ll need to adjust the broker tuning on
> the
> > clusters you are consuming from, since that is the high latency side. The
> > configuration for that is socket.send.buffer.bytes, and it probably makes
> > sense to set this to -1 (which means use the OS configuration). You can
> do
> > the same with socket.receive.buffer.bytes, but it’s not as critical with
> > this setup. On the mirror maker, the configuration is on the consumer
> side,
> > and it’s called receive.buffer.bytes. Again, you can set this to -1 to
> use
> > the OS configuration. Make sure to restart the applications after making
> > all these changes, of course.
> >
> > -Todd
> >
> >
> > On Sat, Jul 22, 2017 at 1:27 AM, James Cheng <wushuja...@gmail.com>
> wrote:
> >
> >> Becket Qin from LinkedIn spoke at a meetup about how to tune the Kafka
> >> producer. One scenario that he described was tuning for situations where
> >> you had high network latency. See slides at https://www.slideshare.net/
> >> mobile/JiangjieQin/producer-performance-tuning-for-apache-
> kafka-63147600
> >> and video at https://youtu.be/oQe7PpDDdzA
> >>
> >> -James
> >>
> >> Sent from my iPhone
> >>
> >>> On Jul 21, 2017, at 9:25 AM, Sunil Parmar <sunilosu...@gmail.com>
> wrote:
> >>>
> >>> We're trying to set up mirror maker to mirror data from EU dc to US dc.
> >> The
> >>> network delay is ~150 ms. In recent test; we realized that mirror maker
> >> is
> >>> not keeping up with load and have a lag trending upward all the time.
> >>>
> >>> What are configurations that can be tuned up to make it work for the
> >> higher
> >>> throughput ?
> >>> How to decide number of producer and consumer threads ? ( number of
> topic
> >>> partitions / brokers ? )
> >>>
> >>>
> >>> *Environment* ( both source and destination cluster )
> >>>
> >>> Kafka version 0.9 ( Cloudera 0.9.0.0+kafka2.0.0+188 )
> >>>
> >>> queue.size = 1
> >>> queue.byte.size = 100MB
> >>>
> >>> 2 brokers on source, 3 brokers on destination
> >>>
> >>>
> >>> *Mirror maker configs :*
> >>>
> >>> Producer properties :
> >>> request.timeout.ms=12
> >>> block.on.buffer.full=TRUE
> >>> retries=20
> >>> acks=all
> >>>
> >>>
> >>> Consumer properties:
> >>> request

Re: Tuning up mirror maker for high thruput

[Todd Palino]

One of the best pieces of advice I can offer is that you really need to run
the mirror maker in the same physical/network location as the Kafka cluster
you are producing to. Latency on the consumer side can be more easily
absorbed than latency on the producer side, as to assure that we have
proper message ordering and reliability, we need to restrict in flight
batches to 1. So that means that our produce connection is contstrained to
be very thin, and latency makes a huge impact. Meanwhile, on the consume
side we’re fetching large batches of messages, many at a time, so
round-trip latency has less of an impact. I really can’t stress this
enough. We set up some mirror makers in the opposite configuration for
security reasons, and it’s been a huge problem even with tuning.

In addition to this, you will want to assure that your OS (and then the
mirror maker and broker) tuning is taking into account the latency. Here’s
a good reference for the OS side (for Linux):
http://linuxczar.net/blog/2016/09/18/bandwidth-delay-product/

Once you have the OS tuned, you’ll need to adjust the broker tuning on the
clusters you are consuming from, since that is the high latency side. The
configuration for that is socket.send.buffer.bytes, and it probably makes
sense to set this to -1 (which means use the OS configuration). You can do
the same with socket.receive.buffer.bytes, but it’s not as critical with
this setup. On the mirror maker, the configuration is on the consumer side,
and it’s called receive.buffer.bytes. Again, you can set this to -1 to use
the OS configuration. Make sure to restart the applications after making
all these changes, of course.

-Todd


On Sat, Jul 22, 2017 at 1:27 AM, James Cheng <wushuja...@gmail.com> wrote:

> Becket Qin from LinkedIn spoke at a meetup about how to tune the Kafka
> producer. One scenario that he described was tuning for situations where
> you had high network latency. See slides at https://www.slideshare.net/
> mobile/JiangjieQin/producer-performance-tuning-for-apache-kafka-63147600
> and video at https://youtu.be/oQe7PpDDdzA
>
> -James
>
> Sent from my iPhone
>
> > On Jul 21, 2017, at 9:25 AM, Sunil Parmar <sunilosu...@gmail.com> wrote:
> >
> > We're trying to set up mirror maker to mirror data from EU dc to US dc.
> The
> > network delay is ~150 ms. In recent test; we realized that mirror maker
> is
> > not keeping up with load and have a lag trending upward all the time.
> >
> > What are configurations that can be tuned up to make it work for the
> higher
> > throughput ?
> > How to decide number of producer and consumer threads ? ( number of topic
> > partitions / brokers ? )
> >
> >
> > *Environment* ( both source and destination cluster )
> >
> > Kafka version 0.9 ( Cloudera 0.9.0.0+kafka2.0.0+188 )
> >
> > queue.size = 1
> > queue.byte.size = 100MB
> >
> > 2 brokers on source, 3 brokers on destination
> >
> >
> > *Mirror maker configs :*
> >
> > Producer properties :
> > request.timeout.ms=12
> > block.on.buffer.full=TRUE
> > retries=20
> > acks=all
> >
> >
> > Consumer properties:
> > request.timeout.ms=12
> > auto.offset.reset=latest
> > enable.auto.commit=false
> >
> > We've configured 4 producer and consumer threads.
> > There is no security set up as of now so it's all PLAINTEXT.
> >
> > We have 4 topics are white listed to sync from EU to US. Only one of them
> > is high throughput. We also have a message handler to strip off some
> > sensitive information from EU to US but it only works on a low thru put
> > topic; the message handler still try to process the other topics but let
> it
> > pass thru.
> >
> > Thanks,
> > Sunil Parmar
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka Monitoring

[Todd Palino]

Not for monitoring Kafka. We pull the JMX metrics two ways - one is a
container that wraps around the Kafka application and annotates the beans
to be emitted to Kafka as metrics, which gets pulled into our
autometrics/InGraphs system for graphing. But for alerting, we use an agent
that polls the critical metrics via JMX and pushes them into a separate
system (that doesn’t use Kafka). ELK is used for log analysis for other
applications.

Kafka-monitor is what we built/use for synthetic traffic monitoring for
availability. And Burrow for monitoring consumers.

-Todd


On Tue, Jun 20, 2017 at 9:53 AM, Andrew Hoblitzell <
ahoblitz...@salesforce.com> wrote:

> Using Elasticsearch, Logstash, and Kibana is a pretty popular pattern at
> LinkedIn.
>
> Also giving honorable mentions to Kafka Monitor and Kafka Manager since
> they hadn't been mentioned yet
> https://github.com/yahoo/kafka-manager
> https://github.com/linkedin/kafka-monitor
>
> Thanks,
>
> Andrew Hoblitzell
> Sr. Software Engineer, Salesforce
>
>
> On Tue, Jun 20, 2017 at 9:37 AM, Todd S <t...@borked.ca> wrote:
>
> > You can look at enabling JMX on kafka (
> > https://stackoverflow.com/questions/36708384/enable-jmx-on-kafka-brokers
> )
> > using
> > JMXTrans (https://github.com/jmxtrans/jmxtrans) and a config (
> > https://github.com/wikimedia/puppet-kafka/blob/master/
> > kafka-jmxtrans.json.md)
> > to gather stats, and insert them into influxdb (
> > https://www.digitalocean.com/community/tutorials/how-to-
> > monitor-system-metrics-with-the-tick-stack-on-centos-7)
> > then graph the resulsts with grafana (
> > https://softwaremill.com/monitoring-apache-kafka-with-influxdb-grafana/,
> > https://grafana.com/dashboards/721)
> >
> > This is likely a solid day of work to get working nicely, but it also
> > enables you to do a lot of extra cool stuff for monitoring, more than
> just
> > Kafka.  JMXTrans can be a bit of a pain, because Kafkas JMX metrics are
> ..
> > plentiful ... but the example configuration above should get you started.
> > Using Telegraf to collect system stats and graph them with Grafana is
> > really simple and powerful, as the Grafana community has a lot of
> pre-built
> > content you can steal and make quick wins with.
> >
> > Monitoring Kafka can be a beast, but there is a lot of useful data there
> > for if(when?) there is a problem.  The more time you spend with the
> > metrics, the more you start to get a feel for the internals.
> >
> > On Mon, Jun 19, 2017 at 6:52 PM, Muhammad Arshad <
> > muhammad.ars...@alticeusa.com> wrote:
> >
> > > Hi,
> > >
> > > wanted to see if there is Kafka monitoring which is available. I am
> > > looking to the following:
> > >
> > >
> > >
> > > how much data came in at a certain time.
> > >
> > >
> > >
> > > Thanks,
> > >
> > > *Muhammad Faisal Arshad*
> > >
> > > Manager, Enterprise Data Quality
> > >
> > > Data Services & Architecture
> > >
> > > [image:
> > > http://www.multichannel.com/sites/default/files/public/
> > styles/blog_content/public/Altice-NewLogo2017_RESIZED_0.
> jpg?itok=RmwvsCI6]
> > >
> > >
> > >
> > >
> > > 
> > > The information transmitted in this email and any of its attachments is
> > > intended only for the person or entity to which it is addressed and may
> > > contain information concerning Altice USA and/or its affiliates and
> > > subsidiaries that is proprietary, privileged, confidential and/or
> subject
> > > to copyright. Any review, retransmission, dissemination or other use
> of,
> > or
> > > taking of any action in reliance upon, this information by persons or
> > > entities other than the intended recipient(s) is prohibited and may be
> > > unlawful. If you received this in error, please contact the sender
> > > immediately and delete and destroy the communication and all of the
> > > attachments you have received and all copies thereof.
> > > 
> > >
> > >
> >
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Re-Balancing Kafka topics - Best practices

[Todd Palino]

A few things here…

1) auto.leader.rebalance.enable can have serious performance impacts on
larger clusters. It’s currently in need of some development work to enable
it to batch leader elections into smaller groups and back off between them,
as well as have a better backoff after broker startup. I don’t recommend
using it.

2) auto.leader.rebalance.enable is not going to get you what you’re looking
for. It only changes the leader for a partition to the “optimal” leader (I
put that in quotes because it’s a pretty dumb algorithm. It’s whichever
replica is listed first). It does not move partitions around to assure you
have a balance of traffic across the cluster.

If you want to rebalance partitions, you have a couple options right now:
1) Run kafka-reassign-partitions.sh. It will move all of the partitions
around and try and assure an even count on each broker. It does not balance
traffic, however, (if you have a really busy partition and a really slow
partition, it considers them equal).
2) Use an external tool like https://github.com/linkedin/kafka-tools
kafka-assigner. This is a script we developed at LinkedIn for doing
operations that involve moving partitions around and provides a number of
different ways to rebalance traffic.

There are other tools available for doing this, but right now it requires
something external to the Apache Kafka core.

-Todd


On Tue, Jun 13, 2017 at 5:30 PM, karan alang <karan.al...@gmail.com> wrote:

> Hi All,
>
> Fpr Re-balancing Kafka partitions, we can set property ->
>
>
> *auto.leader.rebalance.enable = true in server.properties file.*
>
> *Is that the recommended way or is it better to reBalance the kafka
> partitions manually ?(using *scripts - *kafka-preferred-replica-
> election.sh,
> *
>
> *kafka-reassign-partition.sh)*
> *One of the blogs mentioned that - it is preferable to Re-balance Kafka
> topics manually, since setting   *
>
> *auto.leader.rebalance.enable = true causes issues.*
>
> Pls let me know.
> Any other best practices wrt. Re-balancing kafka topics ?
>
> thanks!
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Queries regarding kafka Monitoring tool burrow.

[Todd Palino]

ot;offset": 144,
> > "timestamp": 1494566999000,
> > "lag": 0,
> > "max_offset": 144
> > }
> > }
> > ],
> > "partition_count": 17,
> > "maxlag": null,
> > "totallag": 0
> > },
> > "request": {
> > "url": "/v2/kafka/local/consumer/group1/lag",
> > "host": "",
> > "cluster": "local",
> > "group": "group1",
> > "topic": ""
> > }
> > }
> >
> >
> > 2. Since burrow returns JSON is there any visualization tools that can be
> > used to monitor the end results.
> >
> > 3. On hitting group describe command and burrow group lag command results
> > are different burrow result is somewhat delayed then the results that I
> am
> > getting while hitting group describes command on Kafka broker and I am
> > getting the different result.
> >
> >
> >
> > Below mentioned is my burrow configuration:
> >
> >
> > [general]
> > logdir=log
> > logconfig=/root/go/src/github.com/linkedin/Burrow/config/logging.cfg
> > pidfile=burrow.pid
> > client-id=burrow-lagchecker
> > group-blacklist=^(console-consumer-|python-kafka-consumer-).*$
> > #group-whitelist=^(my-important-consumer).*$
> >
> > [zookeeper]
> > hostname=
> > port=2181
> > timeout=6
> > lock-path=/burrow/notifier
> >
> > [kafka "local"]
> > broker=
> > broker-port=9092
> > zookeeper=
> > zookeeper-port=2181
> > zookeeper-path=/
> > offsets-topic=__consumer_offsets
> >
> > #[storm "local"]
> > #zookeeper=zkhost01.example.com
> > #zookeeper-port=2181
> > #zookeeper-path=/kafka-cluster/stormconsumers
> >
> > [tickers]
> > broker-offsets=20
> >
> > [lagcheck]
> > intervals=10
> > expire-group=604800
> >
> > [notify]
> > interval=10
> >
> > [httpserver]
> > server=on
> > port=8000
> > ; Alternatively, use listen (cannot be specified when port is)
> > ; listen=host:port
> > ; listen=host2:port2
> >
> > [smtp]
> > server=mailserver.example.com
> > port=25
> > from=burrow-nore...@example.com
> > template=config/default-email.tmpl
> >
> > [emailnotifier "b...@example.com"]
> > group=local,critical-consumer-group
> > group=local,other-consumer-group
> > interval=60
> >
> > [notify]
> > interval=10
> >
> > [httpnotifier]
> > url=http://notification.server.example.com:9000/v1/alert
> > interval=60
> > extra=app=burrow
> > extra=tier=STG
> > template-post=config/default-http-post.tmpl
> > template-delete=config/default-http-delete.tmpl
> > timeout=5
> > keepalive=30
> >
> > So Can you please let me know what I am missing and how to fix these
> > issues.Any help would be appreciated.
> >
> >
> >
> > Regards,
> > Abhimanyu
> >
>



-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: How to increase replication factor in kafka 10.2

[Todd Palino]

You can use the reassign partitions CLI tool to generate a partition
reassignment for the topic, and then manually edit the JSON to add a third
replica ID to each partition before you run it.

Alternately, you can use our kafka-assigner tool (
https://github.com/linkedin/kafka-tools) to do it in a more automated
fashion.

-Todd

On Wed, Apr 26, 2017 at 12:25 PM Naanu Bora <naanub...@gmail.com> wrote:

> Hi,
>In our team some developers created topics with replication factor as 1
> by mistake and number of partition in the range of 20-40. How to increase
> the replication factor to 3 for those topics now? Do we need to come up
> with a manual assignment plan for each of the partitions? Is there any
> quicker way to achieve this?
>
> Thanks!
>
-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Elegant way to failover controller

[Todd Palino]

You can force the controller to move, but you can’t control where it goes.
To force the controller to move to another broker, delete the “/controller”
znode in Zookeeper for the cluster. This will cause the current controller
to resign and a new one to be elected, but it’s random as to what broker
that will be.

A better question here is why do you want to move the controller?

-Todd

On Wed, Apr 5, 2017 at 9:09 AM, Jun MA <mj.saber1...@gmail.com> wrote:

> Hi,
>
> We are running kafka 0.9.0.1 and I’m looking for a elegant way to failover
> controller to other brokers. Right now I have to restart the broker to
> failover it to other brokers. Is there a way to failover controller to a
> specific broker? Is there a way to failover it without restart the broker?
>
> Thanks,
> Jun




-- 
*Todd Palino*
Senior Staff Engineer, Site Reliability
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Performance and encryption

[Todd Palino]

They are defined at the broker level as a default for all topics that do
not have an override for those configs. Both (and many other configs) can
be overridden for individual topics using the command line tools.

-Todd


On Wed, Mar 8, 2017 at 12:36 PM, Nicolas Motte <lingusi...@gmail.com> wrote:

> Hi everyone, I have another question.
> Is there any reason why retention and cleanup policy are defined at cluster
> level and not topic level?
> I can t see why it would not be possible from a technical point of view...
>
> 2017-03-06 14:38 GMT+01:00 Nicolas Motte <lingusi...@gmail.com>:
>
> > Hi everyone,
> >
> > I understand one of the reasons why Kafka is performant is by using
> > zero-copy.
> >
> > I often hear that when encryption is enabled, then Kafka has to copy the
> > data in user space to decode the message, so it has a big impact on
> > performance.
> >
> > If it is true, I don t get why the message has to be decoded by Kafka. I
> > would assume that whether the message is encrypted or not, Kafka simply
> > receives it, appends it to the file, and when a consumer wants to read
> it,
> > it simply reads at the right offset...
> >
> > Also I m wondering if it s the case if we don t use keys (pure queuing
> > system with key=null).
> >
> > Cheers
> > Nico
> >
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Performance and Encryption

[Todd Palino]

Nicholas, this appears to be a duplicate of your question from 2 days ago.
Please review that for discussion on this question.

-Todd


On Wed, Mar 8, 2017 at 1:08 PM, Tauzell, Dave <dave.tauz...@surescripts.com>
wrote:

> I think because the product batches messages which could be for different
> topics.
>
> -Dave
>
> -Original Message-
> From: Nicolas MOTTE [mailto:nicolas.mo...@amadeus.com]
> Sent: Wednesday, March 8, 2017 2:41 PM
> To: users@kafka.apache.org
> Subject: Performance and Encryption
>
> Hi everyone,
>
> I understand one of the reasons why Kafka is performant is by using
> zero-copy.
>
> I often hear that when encryption is enabled, then Kafka has to copy the
> data in user space to decode the message, so it has a big impact on
> performance.
>
> If it is true, I don t get why the message has to be decoded by Kafka. I
> would assume that whether the message is encrypted or not, Kafka simply
> receives it, appends it to the file, and when a consumer wants to read it,
> it simply reads at the right offset...
>
> Also I m wondering if it s the case if we don t use keys (pure queuing
> system with key=null).
>
> Cheers
> Nico
>
> This e-mail and any files transmitted with it are confidential, may
> contain sensitive information, and are intended solely for the use of the
> individual or entity to whom they are addressed. If you have received this
> e-mail in error, please notify the sender by reply e-mail immediately and
> destroy all copies of the e-mail and any attachments.
>
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: MirrorMaker and producers

[Todd Palino]

For this type of use case, there’s no problem with mirroring and producing
into the same topic. Kafka can handle it just fine, and as long as you’re
OK with the intermingled data from the consumer side (for example, knowing
that it may not be time-ordered if you’re working with keyed data), it will
work properly.

I’ve often railed against producing into clusters and topics that you are
mirroring into, but this is largely because those clusters are designed to
be aggregates of other clusters. If you produce directly to the aggregate
cluster, it no longer matches the other aggregate clusters. But the
migration use case is different than that.

-Todd


On Mon, Mar 6, 2017 at 2:50 PM, Jack Foy <j...@hiya.com> wrote:

> Hey, all. Is there any general guidance around using mirrored topics
> in the context of a cluster migration?
>
> We're moving operations from one data center to another, and we want
> to stream mirrored data from the old cluster to the new, migrate
> consumers, then migrate producers.
>
> Our basic question is whether it's safe for us to commingle mirrored
> and directly-produced data in the same topic, even serially. In other
> words, is the following procedure safe? Why or why not?
>
> - Data is produced to topic T on cluster A
> - Topic T is mirrored to cluster B
> - Consumers run against T on cluster B
> - Producers gradually migrate from A to B
>
> We've found the following, which seems to suggest no, but doesn't
> address the point directly:
> http://events.linuxfoundation.org/sites/events/files/slides/
> Kafka%20At%20Scale.pdf
>
> --
> Jack Foy <j...@hiya.com>
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Clarification on min.insync.replicas​

[Todd Palino]

It sounds like you’re running into a problem, but we use
min.insync.replicas here and it works. When you describe the topic using
kafka-topics.sh, what does it show you?

On Tue, Mar 7, 2017 at 12:18 PM, James Cheng <wushuja...@gmail.com> wrote:

>
> > On Mar 7, 2017, at 7:44 AM, Shrikant Patel <spa...@pdxinc.com> wrote:
> >
> > Thanks for clarification. I am seeing strange behavior in that case,
> >
> > When I set min.insync.replicas=2 in my server.properties (restart the
> server) and set the acks=all on producer, I am still able to publish to
> topic even when only leader is up (none of followers are alive). With this
> configuration I was hoping to see NotEnoughReplicasException.
> >
> > When I set min.insync.replicas=2 specifically on my topic and set the
> acks=all on producer, I get error NotEnoughReplicasException when less than
> 2 replicas are live.
> >
> > Because of this behavior I thought min.insync.replicas=2 in my
> server.properties does not work and we needed it at explicit topic level.
> >
> > This looks like bug?? Anyone facing this issue.
> >
>
> Did you try min.insync.replicas=2 in server.properties and acks=all on
> producer? min.insync.replicas only applies when acks=all.
>
> -James
>
> >
> > -Original Message-
> > From: Todd Palino [mailto:tpal...@gmail.com]
> > Sent: Monday, March 06, 2017 6:48 PM
> > To: users@kafka.apache.org
> > Subject: Re: Clarification on min.insync.replicas​
> >
> > Default broker configurations do not show in the topic overrides (which
> is what you are showing with the topics tool). It is more accurate to say
> that the min.insync.replicas setting in your server.properties file is what
> will apply to every topic (regardless of when it is created), if there
> exists no topic override for that configuration for that config.
> >
> > -Todd
> >
> >
> > On Mon, Mar 6, 2017 at 4:38 PM, Shrikant Patel <spa...@pdxinc.com>
> wrote:
> >
> >> Hi All,
> >>
> >> Need details about min.insync.replicas​ in the server.properties.
> >>
> >> I thought once I add this to server.properties, all subsequent topic
> >> create should have this as default value.
> >>
> >> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topic
> >> s.bat --zookeeper localhost:2181/chroot/cluster1 --create --topic test
> >> --partition 3 --replication-factor 3 Created topic "test".
> >>
> >> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topic
> >> s.bat --zookeeper localhost:2181/chroot/cluster1 --describe --topic
> >> test
> >> Topic:test  PartitionCount:3ReplicationFactor:3 Configs:
> >>
> >> No min.insync.replicas is set on the topic.
> >>
> >> Why do I have explicit provide this configuratoin when creating topic?
> >> So whats the purpose of this in server.properties??
> >>
> >> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topic
> >> s.bat --zookeeper localhost:2181/chroot/cluster1 --create --topic test
> >> --partition 3 --replication-factor 3 --config min.insync.replicas=3
> >> Created topic "test".
> >>
> >> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topic
> >> s.bat --zookeeper localhost:2181/chroot/cluster1 --describe --topic
> >> test
> >> Topic:test  PartitionCount:3ReplicationFactor:3
> >> Configs:min.insync.replicas=3
> >>Topic: test Partition: 0Leader: 1   Replicas: 1,2,0
> >> Isr: 1,2,0
> >>Topic: test Partition: 1Leader: 2   Replicas: 2,0,1
> >> Isr: 2,0,1
> >>Topic: test Partition: 2Leader: 0   Replicas: 0,1,2
> >> Isr: 0,1,2
> >>
> >> Thanks
> >> Shri
> >>
> >>
> >>
> >> This e-mail and its contents (to include attachments) are the property
> >> of National Health Systems, Inc., its subsidiaries and affiliates,
> >> including but not limited to Rx.com Community Healthcare Network, Inc.
> >> and its subsidiaries, and may contain confidential and proprietary or
> >> privileged information. If you are not the intended recipient of this
> >> e-mail, you are hereby notified that any unauthorized disclosure,
> >> copying, or distribution of this e-mail or of its attachments, or the
> >> taking of any unauthorized action based on information contained herein
> is strictly prohibited.
> >> Unauthorize

Re: Clarification on min.insync.replicas​

[Todd Palino]

Default broker configurations do not show in the topic overrides (which is
what you are showing with the topics tool). It is more accurate to say that
the min.insync.replicas setting in your server.properties file is what will
apply to every topic (regardless of when it is created), if there exists no
topic override for that configuration for that config.

-Todd


On Mon, Mar 6, 2017 at 4:38 PM, Shrikant Patel <spa...@pdxinc.com> wrote:

> Hi All,
>
> Need details about min.insync.replicas​ in the server.properties.
>
> I thought once I add this to server.properties, all subsequent topic
> create should have this as default value.
>
> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topics.bat
> --zookeeper localhost:2181/chroot/cluster1 --create --topic test
> --partition 3 --replication-factor 3
> Created topic "test".
>
> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topics.bat
> --zookeeper localhost:2181/chroot/cluster1 --describe --topic test
> Topic:test  PartitionCount:3ReplicationFactor:3 Configs:
>
> No min.insync.replicas is set on the topic.
>
> Why do I have explicit provide this configuratoin when creating topic? So
> whats the purpose of this in server.properties??
>
> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topics.bat
> --zookeeper localhost:2181/chroot/cluster1 --create --topic test
> --partition 3 --replication-factor 3 --config min.insync.replicas=3
> Created topic "test".
>
> C:\JAVA_INSTALLATION\kafka\kafka_2.11-0.10.1.1>bin\windows\kafka-topics.bat
> --zookeeper localhost:2181/chroot/cluster1 --describe --topic test
> Topic:test  PartitionCount:3ReplicationFactor:3
>  Configs:min.insync.replicas=3
> Topic: test Partition: 0Leader: 1   Replicas: 1,2,0
> Isr: 1,2,0
> Topic: test Partition: 1Leader: 2   Replicas: 2,0,1
> Isr: 2,0,1
> Topic: test Partition: 2Leader: 0   Replicas: 0,1,2
> Isr: 0,1,2
>
> Thanks
> Shri
>
>
>
> This e-mail and its contents (to include attachments) are the property of
> National Health Systems, Inc., its subsidiaries and affiliates, including
> but not limited to Rx.com Community Healthcare Network, Inc. and its
> subsidiaries, and may contain confidential and proprietary or privileged
> information. If you are not the intended recipient of this e-mail, you are
> hereby notified that any unauthorized disclosure, copying, or distribution
> of this e-mail or of its attachments, or the taking of any unauthorized
> action based on information contained herein is strictly prohibited.
> Unauthorized use of information contained herein may subject you to civil
> and criminal prosecution and penalties. If you are not the intended
> recipient, please immediately notify the sender by telephone at
> 800-433-5719 or return e-mail and permanently delete the original e-mail.
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Performance and encryption

[Todd Palino]

Thanks for the link, Ismael. I had thought that the most recent kernels
already implemented this, but I was probably confusing it with BSD. Most of
my systems are stuck in the stone age right now anyway.

It would be nice to get KAFKA-2561 in, either way. First off, if you can
take advantage of it it’s a good performance boost. Second, especially with
the security landscape getting worse and worse, it would be good to have
options as far as the TLS implementation goes. A zero-day exploit in the
Java TLS implementation would be devastating, and more difficult to react
to as it would require a new JRE (bringing with it who knows what
problems). Swapping an underlying OpenSSL version would be much more
palatable.

-Todd


On Mon, Mar 6, 2017 at 9:01 AM, Ismael Juma <ism...@juma.me.uk> wrote:

> Even though OpenSSL is much faster than the Java 8 TLS implementation (I
> haven't tested against Java 9, which is much faster than Java 8, but
> probably still slower than OpenSSL), all the tests were without zero copy
> in the sense that is being discussed here (i.e. sendfile). To benefit from
> sendfile with TLS, kernel-level changes/modules are required:
>
> https://github.com/ktls/af_ktls
> http://www.phoronix.com/scan.php?page=news_item=FreeBSD-Faster-Sendfile
>
> Ismael
>
> On Mon, Mar 6, 2017 at 4:18 PM, Todd Palino <tpal...@gmail.com> wrote:
>
> > So that’s not quite true, Hans. First, as far as the performance hit
> being
> > not a big impact (25% is huge). Or that it’s to be expected. Part of the
> > problem is that the Java TLS implementation does not support zero copy.
> > OpenSSL does, and in fact there’s been a ticket open to allow Kafka to
> > support using OpenSSL for a while now:
> >
> > https://issues.apache.org/jira/browse/KAFKA-2561
> >
> >
> >
> >
> > On Mon, Mar 6, 2017 at 6:30 AM, Hans Jespersen <h...@confluent.io>
> wrote:
> >
> > >
> > > Its not a single message at a time that is encrypted with TLS its the
> > > entire network byte stream so a Kafka broker can’t even see the Kafka
> > > Protocol tunneled inside TLS unless it’s terminated at the broker.
> > > It is true that losing the zero copy optimization impacts performance
> > > somewhat  but it’s not what I would call a “big impact” because Kafka
> > does
> > > a lot of other things to get it’s performance (like using page cache
> and
> > > doing lots on sequential disk I/O). The difference should be something
> in
> > > the order of 25-30% slower with TLS enabled which is about what you
> would
> > > see with any other messaging protocol with TLS on vs off.
> > >
> > > If you wanted to encrypt each message independently before sending to
> > > Kafka then zero copy would still be in effect and all the consumers
> would
> > > get the same encrypted message (and have to understand how to decrypt
> > it).
> > >
> > > -hans
> > >
> > >
> > >
> > > > On Mar 6, 2017, at 5:38 AM, Nicolas Motte <lingusi...@gmail.com>
> > wrote:
> > > >
> > > > Hi everyone,
> > > >
> > > > I understand one of the reasons why Kafka is performant is by using
> > > > zero-copy.
> > > >
> > > > I often hear that when encryption is enabled, then Kafka has to copy
> > the
> > > > data in user space to decode the message, so it has a big impact on
> > > > performance.
> > > >
> > > > If it is true, I don t get why the message has to be decoded by
> Kafka.
> > I
> > > > would assume that whether the message is encrypted or not, Kafka
> simply
> > > > receives it, appends it to the file, and when a consumer wants to
> read
> > > it,
> > > > it simply reads at the right offset...
> > > >
> > > > Also I m wondering if it s the case if we don t use keys (pure
> queuing
> > > > system with key=null).
> > > >
> > > > Cheers
> > > > Nico
> > >
> > >
> >
> >
> > --
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Performance and encryption

[Todd Palino]

So that’s not quite true, Hans. First, as far as the performance hit being
not a big impact (25% is huge). Or that it’s to be expected. Part of the
problem is that the Java TLS implementation does not support zero copy.
OpenSSL does, and in fact there’s been a ticket open to allow Kafka to
support using OpenSSL for a while now:

https://issues.apache.org/jira/browse/KAFKA-2561




On Mon, Mar 6, 2017 at 6:30 AM, Hans Jespersen <h...@confluent.io> wrote:

>
> Its not a single message at a time that is encrypted with TLS its the
> entire network byte stream so a Kafka broker can’t even see the Kafka
> Protocol tunneled inside TLS unless it’s terminated at the broker.
> It is true that losing the zero copy optimization impacts performance
> somewhat  but it’s not what I would call a “big impact” because Kafka does
> a lot of other things to get it’s performance (like using page cache and
> doing lots on sequential disk I/O). The difference should be something in
> the order of 25-30% slower with TLS enabled which is about what you would
> see with any other messaging protocol with TLS on vs off.
>
> If you wanted to encrypt each message independently before sending to
> Kafka then zero copy would still be in effect and all the consumers would
> get the same encrypted message (and have to understand how to decrypt it).
>
> -hans
>
>
>
> > On Mar 6, 2017, at 5:38 AM, Nicolas Motte <lingusi...@gmail.com> wrote:
> >
> > Hi everyone,
> >
> > I understand one of the reasons why Kafka is performant is by using
> > zero-copy.
> >
> > I often hear that when encryption is enabled, then Kafka has to copy the
> > data in user space to decode the message, so it has a big impact on
> > performance.
> >
> > If it is true, I don t get why the message has to be decoded by Kafka. I
> > would assume that whether the message is encrypted or not, Kafka simply
> > receives it, appends it to the file, and when a consumer wants to read
> it,
> > it simply reads at the right offset...
> >
> > Also I m wondering if it s the case if we don t use keys (pure queuing
> > system with key=null).
> >
> > Cheers
> > Nico
>
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Performance and encryption

[Todd Palino]

It’s not that Kafka has to decode it, it’s that it has to send it across
the network. This is specific to enabling TLS support (transport
encryption), and won’t affect any end-to-end encryption you do at the
client level.

The operation in question is called “zero copy”. In order to send a message
batch to a consumer, the Kafka broker must read it from disk (sometimes
it’s cached in memory, but that’s irrelevant here) and send it across the
network. The Linux kernel allows this to happen without having to copy the
data in memory (to move it from the disk buffers to the network buffers).
However, if TLS is enabled, the broker must first encrypt the data going
across the network. This means that it can no longer take advantage of the
zero copy optimization as it has to make a copy in the process of applying
the TLS encryption.

Now, how much of an impact this has on the broker operations is up for
debate, I think. Originally, when we ran into this problem was when TLS
support was added to Kafka and the zero copy send for plaintext
communications was accidentally removed as well. At the time, we saw a
significant performance hit, and the code was patched to put it back.
However, since then I’ve turned on inter-broker TLS in all of our clusters,
and when we did that there was no performance hit. This is odd, because the
replica fetchers should take advantage of the same zero copy optimization.

It’s possible that it’s because it’s just one consumer (the replica
fetchers). We’re about to start testing additional consumers over TLS, so
we’ll see what happens at that point. All I can suggest right now is that
you test in your environment and see what the impact is. Oh, and using
message keys (or not) won’t matter here.

-Todd


On Mon, Mar 6, 2017 at 5:38 AM, Nicolas Motte <lingusi...@gmail.com> wrote:

> Hi everyone,
>
> I understand one of the reasons why Kafka is performant is by using
> zero-copy.
>
> I often hear that when encryption is enabled, then Kafka has to copy the
> data in user space to decode the message, so it has a big impact on
> performance.
>
> If it is true, I don t get why the message has to be decoded by Kafka. I
> would assume that whether the message is encrypted or not, Kafka simply
> receives it, appends it to the file, and when a consumer wants to read it,
> it simply reads at the right offset...
>
> Also I m wondering if it s the case if we don t use keys (pure queuing
> system with key=null).
>
> Cheers
> Nico
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: How to stop Kafka Mirror Maker

[Todd Palino]

You should just kill the process without the -9. That will send a SIGHUP.
This allows the process to shut down cleanly and flush offset commits
before exiting. Otherwise you’re going to have more duplicate messages (or
dropped messages, depending on your configuration for offset commits). You
should also stop all the mirror maker instances in a given consumer group
in parallel, as this will minimize the number of rebalances and how long it
takes for you to start passing messages again.

-Todd


On Tue, Feb 21, 2017 at 5:14 PM, Qian Zhu <qi...@trulia.com> wrote:

> Hi,
>For now, I am doing “kill –9 processID” to stop the Kafka Mirror
> Maker. I am wondering whether there is a better way (e.g. a command) to do
> so? I don’t expect to stop the mirror maker frequently but I would like to
> have a script to automate the start and stop.
>
> Thanks a lot!
> Qian Zhu
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: hitting the throughput limit on a cluster?

[Todd Palino]

Well, my IOwait sits around 0. Much less than 1% all the time. Any IOwait
at all indicates that the application is waiting on the disk, so that’s
where your bottleneck will be.

As far as open files, it depends on how many log segments and network
connections are open. Because I have large clusters with thousands of
partitions per broker, and tons of network connections from clients, I have
the FD limit set to 400k. Basically, you don’t want to run out, so you want
enough buffer to catch a problem (like a bug with socket file descriptors
not getting released properly).

-Todd


On Tue, Feb 21, 2017 at 12:52 PM, Jon Yeargers <jon.yearg...@cedexis.com>
wrote:

> Thanks for looking at this issue. I checked the max IOPs for this disk and
> we're only at about 10%. I can add more disks to spread out the work.
>
> What IOWait values should I be aiming for?
>
> Also - what do you set openfiles to? I have it at 65535 but I just read a
> doc that suggested > 100K is better
>
>
> On Tue, Feb 21, 2017 at 10:45 AM, Todd Palino <tpal...@gmail.com> wrote:
>
> > So I think the important thing to look at here is the IO wait on your
> > system. You’re hitting disk throughput issues, and that’s what you most
> > likely need to resolve. So just from what you’ve described, I think the
> > only thing that is going to get you more performance is more spindles (or
> > faster spindles). This is either more disks or more brokers, but at the
> end
> > of it you need to eliminate the disk IO bottleneck.
> >
> > -Todd
> >
> >
> > On Tue, Feb 21, 2017 at 7:29 AM, Jon Yeargers <jon.yearg...@cedexis.com>
> > wrote:
> >
> > > Running 3x 8core on google compute.
> > >
> > > Topic has 16 partitions (replication factor 2) and is consumed by 16
> > docker
> > > containers on individual hosts.
> > >
> > > System seems to max out at around 4 messages / minute. Each message
> > is
> > > ~12K - compressed (snappy) JSON.
> > >
> > > Recently moved from 12 to the above 16 partitions with no change in
> > > throughput.
> > >
> > > Also tried increased the consumption capacity on each container by 50%.
> > No
> > > effect.
> > >
> > > Network is running at ~6Gb/sec (measured using iperf3). Broker load is
> > > ~1.5. IOWait % is 5-10 (via sar).
> > >
> > > What are my options for adding throughput?
> > >
> > > - more brokers?
> > > - avro/protobuf messaging?
> > > - more disks / broker? (1 / host presently)
> > > - jumbo frames?
> > >
> > > (transparent huge pages is disabled)
> > >
> > >
> > > Looking at this article (
> > > https://engineering.linkedin.com/kafka/benchmarking-apache-
> > > kafka-2-million-writes-second-three-cheap-machines)
> > > it would appear that for our message size we are at the max. This would
> > > argue that we need to shrink the message size - so perhaps switching to
> > > avro is the next step?
> > >
> >
> >
> >
> > --
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: hitting the throughput limit on a cluster?

[Todd Palino]

So I think the important thing to look at here is the IO wait on your
system. You’re hitting disk throughput issues, and that’s what you most
likely need to resolve. So just from what you’ve described, I think the
only thing that is going to get you more performance is more spindles (or
faster spindles). This is either more disks or more brokers, but at the end
of it you need to eliminate the disk IO bottleneck.

-Todd


On Tue, Feb 21, 2017 at 7:29 AM, Jon Yeargers <jon.yearg...@cedexis.com>
wrote:

> Running 3x 8core on google compute.
>
> Topic has 16 partitions (replication factor 2) and is consumed by 16 docker
> containers on individual hosts.
>
> System seems to max out at around 4 messages / minute. Each message is
> ~12K - compressed (snappy) JSON.
>
> Recently moved from 12 to the above 16 partitions with no change in
> throughput.
>
> Also tried increased the consumption capacity on each container by 50%. No
> effect.
>
> Network is running at ~6Gb/sec (measured using iperf3). Broker load is
> ~1.5. IOWait % is 5-10 (via sar).
>
> What are my options for adding throughput?
>
> - more brokers?
> - avro/protobuf messaging?
> - more disks / broker? (1 / host presently)
> - jumbo frames?
>
> (transparent huge pages is disabled)
>
>
> Looking at this article (
> https://engineering.linkedin.com/kafka/benchmarking-apache-
> kafka-2-million-writes-second-three-cheap-machines)
> it would appear that for our message size we are at the max. This would
> argue that we need to shrink the message size - so perhaps switching to
> avro is the next step?
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: stuck re-balance

[Todd Palino]

Did you move the controller (by deleting the /controller znode) after
removing the reassign_partitions znode? If not, the controller is probably
still trying to do that move, and is not going to accept a new move request.

On Fri, Jan 27, 2017 at 10:16 AM, Tom Raney <tom.ra...@urbanairship.com>
wrote:

> After adding a new Kafka node, I ran the kafka-reassign-partitions.sh tool
> to redistribute topics onto the new machine and it seemed like some of the
> migrations were stuck processing for over 24 hours, so I cancelled the
> reassignment by deleting the zk node (/admin/reassign_partitions) and used
> the kafka-preferred-replica-election.sh to try and resolve it.  It didn't
> work.
>
> Now, I have partitions in a weird state.  For example, I have one partition
> that has broker 1003 as a replica but it shouldn't be there.  The partition
> directory on 1003 is still growing but is way behind the leader and the
> other ISR on 1001.
>
> Topic: foo Partition: 2 Leader: 1004 Replicas: 1003,1004,1001 Isr:
> 1004,1001
>
> When I force a leader election, for that partition, it fails because 1003
> is not in sync.
>
> kafka.common.StateChangeFailedException: encountered error while electing
> leader for partition [foo,2] due to: Preferred replica 1003 for partition
> [foo,2] is either not alive or not in the isr. Current leader and ISR:
> [{"leader":1004,"leader_epoch":11,"isr":[1004,1001]}].
>
> When I try to reassign with the config...
>
> {"version":1,"partitions":[{"topic":"foo","partition":2,"
> replicas":[1004,1001]}]}
>
> I see that it doesn't resolve.
>
> Status of partition reassignment:
> Reassignment of partition [foo,2] is still in progress
>
> And, I would think it would since 1001 is already an ISR and the leader is
> already 1004.
>
> How do I resolve this?
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Handling Leap second delay

[Todd Palino]

Our method of handling the leap second (for everything, not just Kafka) is
to shut off the NTP servers that all the systems refer to before the leap
second, and turn it back on afterwards. The NTP server will then update
from a stratum 1 server and will eventually step the time correctly on all
the servers. It’s worked for the last couple without a problem.

-Todd

On Tue, Dec 20, 2016 at 7:55 PM, Sanjeev T <san...@gmail.com> wrote:

> Hi,
>
> Can some of you share points on, the versions and handling leap second
> delay on Dec 31, 2016.
>
> Regards
> -Sanjeev
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: stunning error - Request of length 1550939497 is not valid, it is larger than the maximum size of 104857600 bytes

[Todd Palino]

Are you actually getting requests that are 1.3 GB in size, or is something
else happening, like someone trying to make HTTP requests against the Kafka
broker port?

-Todd


On Mon, Dec 12, 2016 at 4:19 AM, Ramya Ramamurthy <
ramyaramamur...@teledna.com> wrote:

> We have got exactly the same problem.
> nvalid receive (size = 1347375956 larger than 104857600).
>
> When trying to increase the size, Java Out of Memory Exception.
> Did you find a work around for the same ??
>
> Thanks.
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Deleting a topic without delete.topic.enable=true?

[Todd Palino]

If it issues topic metadata requests to find out what brokers the topic
partitions live on (so it can make specific requests, like offsets, or
getting JMX metrics from those brokers for the partitions), then it could
easily have caused this. We’ve seen this behavior from multiple types of
monitoring applications.

-Todd


On Mon, Dec 12, 2016 at 8:47 AM, Tim Visher <tim.vis...@gmail.com> wrote:

> I wonder if datadog monitoring triggers that behavior. That's the only
> other piece of our infrastructure that may have been talking to that topic.
>
> On Mon, Dec 12, 2016 at 12:40 AM, Surendra , Manchikanti <
> surendra.manchika...@gmail.com> wrote:
>
> > If "auto.create.topics.enable" is set to true in your configurations ,
> any
> > producer/consumer or fetch request will create the topic again. Set it to
> > false and delete the topic.
> >
> > -- Surendra Manchikanti
> >
> > On Sat, Dec 10, 2016 at 10:59 AM, Todd Palino <tpal...@gmail.com> wrote:
> >
> > > Are you running something else besides the consumers that would
> maintain
> > a
> > > memory of the topics and potentially recreate them by issuing a
> metadata
> > > request? For example, Burrow (the consumer monitoring app I wrote) does
> > > this because it maintains a list of all topics in memory, and will end
> up
> > > recreating a topic that has been deleted as it issues a metadata
> request
> > to
> > > try and find out what happened after an offset request for the topic
> > fails.
> > >
> > > -Todd
> > >
> > >
> > > On Fri, Dec 9, 2016 at 8:37 AM, Tim Visher <tim.vis...@gmail.com>
> wrote:
> > >
> > > > On Fri, Dec 9, 2016 at 11:34 AM, Todd Palino <tpal...@gmail.com>
> > wrote:
> > > >
> > > > > Given that you said you removed the log directories, and provided
> > that
> > > > when
> > > > > you did the rmr on Zookeeper it was to the “/brokers/topics/(topic
> > > name)”
> > > > > path, you did the right things for a manual deletion. It sounds
> like
> > > you
> > > > > may have a consumer (or other client) that is recreating the topic.
> > Do
> > > > you
> > > > > have auto topic creation enabled?
> > > > >
> > > >
> > > > That was the last epiphany we had. We had shut down the producer but
> > not
> > > > all the consumers and we do allow auto-topic creation.
> > > >
> > > > That said, we then proceeded to shut all of them down (the consumers)
> > and
> > > > the topic came back. I'm glad that we were doing the right steps
> > though.
> > > >
> > > > >
> > > > > -Todd
> > > > >
> > > > >
> > > > > On Fri, Dec 9, 2016 at 8:25 AM, Tim Visher <tim.vis...@gmail.com>
> > > wrote:
> > > > >
> > > > > > I did all of that because setting delete.topic.enable=true wasn't
> > > > > > effective. We set that across every broker, restarted them, and
> > then
> > > > > > deleted the topic, and it was still stuck in existence.
> > > > > >
> > > > > > On Fri, Dec 9, 2016 at 11:11 AM, Ali Akhtar <
> ali.rac...@gmail.com>
> > > > > wrote:
> > > > > >
> > > > > > > You need to also delete / restart zookeeper, its probably
> storing
> > > the
> > > > > > > topics there. (Or yeah, just enable it and then delete the
> topic)
> > > > > > >
> > > > > > > On Fri, Dec 9, 2016 at 9:09 PM, Rodrigo Sandoval <
> > > > > > > rodrigo.madfe...@gmail.com
> > > > > > > > wrote:
> > > > > > >
> > > > > > > > Why did you do all those things instead of just setting
> > > > > > > > delete.topic.enable=true?
> > > > > > > >
> > > > > > > > On Dec 9, 2016 13:40, "Tim Visher" <tim.vis...@gmail.com>
> > wrote:
> > > > > > > >
> > > > > > > > > Hi Everyone,
> > > > > > > > >
> > > > > > > > > I'm really confused at the moment. We created a topic with
> > > > brokers
> > > > > > set
> > > > > > > to
> > > > > > > > > delete.topic.enable=false.
> > > &

Re: Deleting a topic without delete.topic.enable=true?

[Todd Palino]

Are you running something else besides the consumers that would maintain a
memory of the topics and potentially recreate them by issuing a metadata
request? For example, Burrow (the consumer monitoring app I wrote) does
this because it maintains a list of all topics in memory, and will end up
recreating a topic that has been deleted as it issues a metadata request to
try and find out what happened after an offset request for the topic fails.

-Todd


On Fri, Dec 9, 2016 at 8:37 AM, Tim Visher <tim.vis...@gmail.com> wrote:

> On Fri, Dec 9, 2016 at 11:34 AM, Todd Palino <tpal...@gmail.com> wrote:
>
> > Given that you said you removed the log directories, and provided that
> when
> > you did the rmr on Zookeeper it was to the “/brokers/topics/(topic name)”
> > path, you did the right things for a manual deletion. It sounds like you
> > may have a consumer (or other client) that is recreating the topic. Do
> you
> > have auto topic creation enabled?
> >
>
> That was the last epiphany we had. We had shut down the producer but not
> all the consumers and we do allow auto-topic creation.
>
> That said, we then proceeded to shut all of them down (the consumers) and
> the topic came back. I'm glad that we were doing the right steps though.
>
> >
> > -Todd
> >
> >
> > On Fri, Dec 9, 2016 at 8:25 AM, Tim Visher <tim.vis...@gmail.com> wrote:
> >
> > > I did all of that because setting delete.topic.enable=true wasn't
> > > effective. We set that across every broker, restarted them, and then
> > > deleted the topic, and it was still stuck in existence.
> > >
> > > On Fri, Dec 9, 2016 at 11:11 AM, Ali Akhtar <ali.rac...@gmail.com>
> > wrote:
> > >
> > > > You need to also delete / restart zookeeper, its probably storing the
> > > > topics there. (Or yeah, just enable it and then delete the topic)
> > > >
> > > > On Fri, Dec 9, 2016 at 9:09 PM, Rodrigo Sandoval <
> > > > rodrigo.madfe...@gmail.com
> > > > > wrote:
> > > >
> > > > > Why did you do all those things instead of just setting
> > > > > delete.topic.enable=true?
> > > > >
> > > > > On Dec 9, 2016 13:40, "Tim Visher" <tim.vis...@gmail.com> wrote:
> > > > >
> > > > > > Hi Everyone,
> > > > > >
> > > > > > I'm really confused at the moment. We created a topic with
> brokers
> > > set
> > > > to
> > > > > > delete.topic.enable=false.
> > > > > >
> > > > > > We now need to delete that topic. To do that we shut down all the
> > > > > brokers,
> > > > > > deleted everything under log.dirs and logs.dir on all the kafka
> > > > brokers,
> > > > > > `rmr`ed the entire chroot that kafka was storing things under in
> > > > > zookeeper,
> > > > > > and then brought kafka back up.
> > > > > >
> > > > > > After doing all that, the topic comes back, every time.
> > > > > >
> > > > > > What can we do to delete that topic?
> > > > > >
> > > > > > --
> > > > > >
> > > > > > In Christ,
> > > > > >
> > > > > > Timmy V.
> > > > > >
> > > > > > http://blog.twonegatives.com/
> > > > > > http://five.sentenc.es/ -- Spend less time on mail
> > > > > >
> > > > >
> > > >
> > >
> >
> >
> >
> > --
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Deleting a topic without delete.topic.enable=true?

[Todd Palino]

Given that you said you removed the log directories, and provided that when
you did the rmr on Zookeeper it was to the “/brokers/topics/(topic name)”
path, you did the right things for a manual deletion. It sounds like you
may have a consumer (or other client) that is recreating the topic. Do you
have auto topic creation enabled?

-Todd


On Fri, Dec 9, 2016 at 8:25 AM, Tim Visher <tim.vis...@gmail.com> wrote:

> I did all of that because setting delete.topic.enable=true wasn't
> effective. We set that across every broker, restarted them, and then
> deleted the topic, and it was still stuck in existence.
>
> On Fri, Dec 9, 2016 at 11:11 AM, Ali Akhtar <ali.rac...@gmail.com> wrote:
>
> > You need to also delete / restart zookeeper, its probably storing the
> > topics there. (Or yeah, just enable it and then delete the topic)
> >
> > On Fri, Dec 9, 2016 at 9:09 PM, Rodrigo Sandoval <
> > rodrigo.madfe...@gmail.com
> > > wrote:
> >
> > > Why did you do all those things instead of just setting
> > > delete.topic.enable=true?
> > >
> > > On Dec 9, 2016 13:40, "Tim Visher" <tim.vis...@gmail.com> wrote:
> > >
> > > > Hi Everyone,
> > > >
> > > > I'm really confused at the moment. We created a topic with brokers
> set
> > to
> > > > delete.topic.enable=false.
> > > >
> > > > We now need to delete that topic. To do that we shut down all the
> > > brokers,
> > > > deleted everything under log.dirs and logs.dir on all the kafka
> > brokers,
> > > > `rmr`ed the entire chroot that kafka was storing things under in
> > > zookeeper,
> > > > and then brought kafka back up.
> > > >
> > > > After doing all that, the topic comes back, every time.
> > > >
> > > > What can we do to delete that topic?
> > > >
> > > > --
> > > >
> > > > In Christ,
> > > >
> > > > Timmy V.
> > > >
> > > > http://blog.twonegatives.com/
> > > > http://five.sentenc.es/ -- Spend less time on mail
> > > >
> > >
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: When using mirrormaker, how are people creating topics?

[Todd Palino]

For most of our clusters, we just use auto topic creation and it’s handled
that way. Periodically we’ll go through and clean up partition counts
across everything if there’s a new high-volume topic. We also have the
ability for people to pre-create topics using a central management system.

For the special mirror maker that we have that does 1-to-1 mappings between
partitions for clusters that do not have auto topic creation enabled, the
topic creation (or partition count changes) are taken care of in the
message handler.

-Todd


On Mon, Dec 5, 2016 at 12:42 AM, James Cheng <wushuja...@gmail.com> wrote:

> Hi,
>
> We are using mirrormaker to mirror topics from one cluster to another, and
> I wanted to get some advice from the community on how people are doing
> mirroring. In particular, how are people dealing with topic creation?
>
> Do you turn on auto-topic creation in your destination clusters
> (auto.create.topics.enable=true)?
>
> If not, do you manually create the individual destination topics?
>
> If so, how does that work with mirroring based on a whitelist (regex)?
>
> The way we are doing it right now is, we have our regex in a file
> somewhere. The regex is used in 2 ways:
> 1) Passed to mirrormaker, to do the mirroring.
> 2) Passed to a program which looks up all the topics on the source
> cluster, finds the ones that match the regex, and then creates them on the
> destination cluster. (We have auto-topic creation turned off
> auto.create.topics.enable=false)
>
> One downside of that approach is there a potential race, where if the
> regex changes, then mirrormaker (in #1) might start trying to produce to a
> new destination topic before the topic was created (by #2).
>
> Some other hand-wavy ideas that came to mind might be:
> * handling topic creation in a MirrorMakerMessageHandler
> * handling topic creation in an interceptor
>
> Anyway, was hoping to get some thoughts from people who are already doing
> this.
>
> Thanks!
> -James
>
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: How to move a broker out of rotation?

[Todd Palino]

There’s not a good answer for this with just the Kafka tools. We opened
sourced the tool that we use for removing brokers and rebalancing
partitions in a cluster:
https://github.com/linkedin/kafka-tools

So when we want to remove a broker (with an ID of 1 in this example) from a
cluster, we run:
kafka-assigner -z zookeeper.example.com:2181 -e remove -b 1

That runs a bunch of partition reassignments to move all replicas off that
broker and distribute them to the other brokers in the cluster.

-Todd


On Thu, Sep 29, 2016 at 3:53 PM, Praveen <praveev...@gmail.com> wrote:

> I have 16 brokers. Now one of the brokers (B-16) got completely messed up
> and is sent for repair.
>
> But I can still see some partitions including the B-16 in its replicas,
> thereby becoming under-replicated.
>
> Is there a proper way to take broker out of rotation?
>
> Praveen
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Burrow E-Mail to GMail

[Todd Palino]

So I can’t speak for general Gmail, but we have been using it through Gmail
internally here for a while. Just watch out for those rate limits, because
Burrow can get noisy (depending on your clusters and consumers)!

-Todd


On Mon, Aug 1, 2016 at 7:30 AM, Brian Dennis <b...@crossjam.net> wrote:

> Burrow users,
>
> Before I head down the rathole of fighting with SMTP servers, does anyone
> have positive or negative confirmation that Burrow can reliably send e-mail
> notifications to a GMail account? On reading the docs, it seems quite
> possible, but I've been bitten in the past trying to automate
> legitimate mail to GMail. Getting washed out with the spammers is a real
> hazard.
>
> Thanks!
>
>
>
> --
> information informs. analysis enlightens.
>
> Brian M. Dennis, PhD
> 43403 Coton Commons Dr, Leesburg, VA 20176
> e: b...@crossjam.net m: 312.213.3943
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: __consumer_offsets rebalance

[Todd Palino]

It's safe to move the partitions of the offsets topic around. You'll move
the consumer coordinators as you do, however, so the one thing you want to
make sure of, especially running an older version, is that log compaction
is working on your brokers and those partitions have been compacted. The
coordinator needs to bootstrap the topic, and if log compaction is broken
that can take a very long time. During that time, it will return errors to
consumers for offset operations, and that can cause offset resets.

-Todd

On Thursday, July 14, 2016, Anderson Goulart <anderson.goul...@boxever.com>
wrote:

> Hi,
>
> I am running kafka 0.8.2.1 under aws instances with multiple availability
> zones. As we want a rack aware partition replication, we have our own
> partition layout distribution, to make sure all partitions are well
> balanced between nodes, leaders and availability zones.
>
> The problem arises with __consumer_offsets internal topic. In our current
> environment it has 50 partitions and are all under the same AZ, with an
> unbalanced leader (all wrong!)
>
> The question is: should I manually change its partition layout
> distribution as I do for the other topics? Is it safe to reassign the new
> layout for this internal topic, using kafka-reassign-partitions.sh?
>
>
> Thanks, Anderson
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Monitoring offset lag

[Todd Palino]

t;  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-1_/10.100.0.113
> Toms-iMac:betwave-server tomdearman$
> /Users/tomdearman/software/kafka_2.11-0.10.0.0/bin/kafka-consumer-groups.sh
> --new-consumer --bootstrap-server localhost:9092 --describe --group
> voidbridge-oneworks-dummy
> GROUP  TOPIC  PARTITION
> CURRENT-OFFSET  LOG-END-OFFSET  LAG OWNER
> voidbridge-oneworks-dummy  integration-oneworks-dummy 2
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-3_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 7
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-3_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 12
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-3_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 17
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-3_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 4
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-5_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 9
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-5_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 14
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-5_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 19
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-5_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 1
> 3   3   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-2_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 6
> 0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-2_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 11
>  0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-2_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 16
>  0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-2_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 3
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-4_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 8
> unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-4_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 13
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-4_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 18
>  unknown 0   unknown
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-4_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 0
> 1   1   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-1_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 5
> 0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-1_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 10
>  0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-1_/10.100.0.113
> voidbridge-oneworks-dummy  integration-oneworks-dummy 15
>  0   0   0
>  integration-oneworks-dummy-voidbridge-oneworks-dummy-1_/10.100.0.113
>
> > On 8 Jul 2016, at 17:20, Tom Dearman <tom.dear...@gmail.com> wrote:
> >
> > When you say ‘for the first partition’ do you literally mean partition
> zero, or you mean any partition.  It is true that when I had only 1 user
> there were only messages on partition 15 but the second user happened to go
> to partition zero.  Is it the case that partition zero must have a consumer
> commit?
> >
> >> On 8 Jul 2016, at 17:16, Todd Palino <tpal...@gmail.com> wrote:
> >>
> >> If you open up an issue on the project, I'd be happy to dig into this in
> >> more detail if needed. Excluding the ZK offset checking, Burrow doesn't
> >> enumerate consumer groups - it learns a

Re: Monitoring offset lag

[Todd Palino]

If you open up an issue on the project, I'd be happy to dig into this in
more detail if needed. Excluding the ZK offset checking, Burrow doesn't
enumerate consumer groups - it learns about them from offset commits. It
sounds like maybe your consumer had not committed offsets for the first
partition (at least not after Burrow was started).

-Todd

On Friday, July 8, 2016, Tom Dearman <tom.dear...@gmail.com> wrote:

> Todd,
>
> Thanks for that I am taking a look.
>
> Is there a bug whereby if you only have a couple of messages on a topic,
> both with the same key, that burrow doesn’t return correct info.  I was
> finding that http://localhost:8100/v2/kafka/betwave/consumer <
> http://localhost:8100/v2/kafka/betwave/consumer> was returning a message
> with empty consumers until I put on another message with a different key,
> i.e. a minimum of 2 partitions with something in them.  I know this is not
> very like production, but on my local this I was only testing with one user
> so get just one partition filled.
>
> Tom
> > On 6 Jul 2016, at 18:08, Todd Palino <tpal...@gmail.com <javascript:;>>
> wrote:
> >
> > Yeah, I've written dissertations at this point on why MaxLag is flawed.
> We
> > also used to use the offset checker tool, and later something similar
> that
> > was a little easier to slot into our monitoring systems. Problems with
> all
> > of these is why I wrote Burrow (https://github.com/linkedin/Burrow)
> >
> > For more details, you can also check out my blog post on the release:
> >
> https://engineering.linkedin.com/apache-kafka/burrow-kafka-consumer-monitoring-reinvented
> >
> > -Todd
> >
> > On Wednesday, July 6, 2016, Tom Dearman <tom.dear...@gmail.com
> <javascript:;>> wrote:
> >
> >> I recently had a problem on my production which I believe was a
> >> manifestation of the issue kafka-2978 (Topic partition is not sometimes
> >> consumed after rebalancing of consumer group), this is fixed in 0.9.0.1
> and
> >> we will upgrade our client soon.  However, it made me realise that I
> didn’t
> >> have any monitoring set up on this.  The only thing I can find as a
> metric
> >> is the
> >>
> kafka.consumer:type=ConsumerFetcherManager,name=MaxLag,clientId=([-.\w]+),
> >> which, if I understand correctly, is the max lag of any partition that
> that
> >> particular consumer is consuming.
> >> 1. If I had been monitoring this, and if my consumer was suffering from
> >> the issue in kafka-2978, would I actually have been alerted, i.e. since
> the
> >> consumer would think it is consuming correctly would it not have updated
> >> the metric.
> >> 2. There is another way to see offset lag using the command
> >> /usr/bin/kafka-consumer-groups --new-consumer --bootstrap-server
> >> 10.10.1.61:9092 --describe —group consumer_group_name and parsing the
> >> response.  Is it safe or advisable to do this?  I like the fact that it
> >> tells me each partition lag, although it is also not available if no
> >> consumer from the group is currently consuming.
> >> 3. Is there a better way of doing this?
> >
> >
> >
> > --
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
>
>

-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka MirrorMaker and Data encryption

[Todd Palino]

We're not doing it in AWS, but we are doing it for other flows. The key to
not having performance problems is to not consume over SSL. This means
turning the previously recommended architecture around, and having the
mirror makers located in the region that you are consuming from. This way
you can consume plaintext and produce over SSL.

Obviously, encryption via produce still has some performance overhead. But
modern processors have optimized instructions for encryption. And not doing
it over the consume side avoids the hit on the brokers from losing the zero
copy send.

-Todd

On Wednesday, July 6, 2016, Reenal Mahajan <reen...@gmail.com> wrote:

> Hi,
>
> We have local kafka clusters in both us-east and us-west data centers in
> AWS. We want to mirror the data in the other cluster so each region has all
> the data. If we use Kafka MirrorMaker, the remote consumer will be
> consuming data over the internet. Is there a way to encrypt this data
> without turning SSL on. There would be some performance degradation and it
> is not required in other scenarios. Has anyone tried doing cross-regional
> replication in production?
>
> Thanks,
> Reenal
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Last offset in all partitions

[Todd Palino]

Yeah, it reflects the last offset in each partition all the time. The one
thing to be careful of is that it can be emitted from any replica
(whichever is the leader). So our tools that use the information always
look at all brokers and then take the larger non-null value.

We've been using it for a while now, and it seems pretty safe. At least
none of our Kafka developers have complained about us doing it :)

-Todd

On Wednesday, July 6, 2016, Kristoffer Sjögren <sto...@gmail.com> wrote:

> That's awesome! I can see the JMX bean [1] in our current 0.8.2
> brokers and the number seems updated in real time per partition? I
> cannot find any official documentation on it? How safe is it to use?
>
> Also we're planning on upgrading to 0.10.0.0 - is it still available?
>
> Cheers,
> -Kristoffer
>
> [1] kafka.log:type=Log,name=LogEndOffset,topic=xxx,partition=xxx
>
> On Wed, Jul 6, 2016 at 5:36 PM, Todd Palino <tpal...@gmail.com
> <javascript:;>> wrote:
> > We do this through our monitoring agents by pulling it as a metric from
> the
> > LogEndOffset beans. By putting it into our metrics system we get a
> mapping
> > of timestamp to offset for every partition with (currently) 60 second
> > granularity. Useful for offset resets and other tasks.
> >
> > -Todd
> >
> > On Wednesday, July 6, 2016, Kristoffer Sjögren <sto...@gmail.com
> <javascript:;>> wrote:
> >
> >> Hi
> >>
> >> Is there a way to get the last offset written by all partitions of a
> >> topic programmatically using the 0.10.0.0 API?
> >>
> >> At the moment I use KafkaConsumer.seekToEnd as seen in this gist[1]
> >> but maybe there is a better, more efficient, way to do it?
> >>
> >> Cheers,
> >> -Kristoffer
> >>
> >> [1] https://gist.github.com/krisskross/a49e462bedb89505e372672cd81129ab
> >>
> >
> >
> > --
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Now available - kafka-tools from LinkedIn SRE

[Todd Palino]

I’ve mentioned this one a couple times in a talk here or a conversation
there, but we hadn’t officially announced a release yet. Now that we’ve
spent some time refactoring the initial code and adding tests, I’m happy to
announce the release of our kafka-tools repository.

The kafka-tools repository (https://github.com/linkedin/kafka-tools) is
where we are starting to release and collaborate on the tools that the SRE
team at LinkedIn uses for managing Kafka clusters. Initially, this is the
kafka-assigner tool that we use for managing partitions in a cluster,
specifically for removing brokers and rebalancing partitions in a cluster.

As always, these are open source and we encourage contributions and
suggestions from everyone. I encourage everyone to take a look and create
PRs for their own features. There’s a couple things that we’re definitely
still working on right now:

   - Adding documentation for everything, including how modules are built
   - Robustness around plugins (to provide local customizations)
   - New balance features, such as rack aware balancing

If there are questions, please feel free to drop me a note. Or open up an
issue on the project. And I want to thank my colleague Jon Bringhurst
profusely for helping to get the structure around the project and the
documentation cleaned up.

-Todd

-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Monitoring offset lag

[Todd Palino]

Yeah, I've written dissertations at this point on why MaxLag is flawed. We
also used to use the offset checker tool, and later something similar that
was a little easier to slot into our monitoring systems. Problems with all
of these is why I wrote Burrow (https://github.com/linkedin/Burrow)

For more details, you can also check out my blog post on the release:
https://engineering.linkedin.com/apache-kafka/burrow-kafka-consumer-monitoring-reinvented

-Todd

On Wednesday, July 6, 2016, Tom Dearman <tom.dear...@gmail.com> wrote:

> I recently had a problem on my production which I believe was a
> manifestation of the issue kafka-2978 (Topic partition is not sometimes
> consumed after rebalancing of consumer group), this is fixed in 0.9.0.1 and
> we will upgrade our client soon.  However, it made me realise that I didn’t
> have any monitoring set up on this.  The only thing I can find as a metric
> is the
> kafka.consumer:type=ConsumerFetcherManager,name=MaxLag,clientId=([-.\w]+),
> which, if I understand correctly, is the max lag of any partition that that
> particular consumer is consuming.
> 1. If I had been monitoring this, and if my consumer was suffering from
> the issue in kafka-2978, would I actually have been alerted, i.e. since the
> consumer would think it is consuming correctly would it not have updated
> the metric.
> 2. There is another way to see offset lag using the command
> /usr/bin/kafka-consumer-groups --new-consumer --bootstrap-server
> 10.10.1.61:9092 --describe —group consumer_group_name and parsing the
> response.  Is it safe or advisable to do this?  I like the fact that it
> tells me each partition lag, although it is also not available if no
> consumer from the group is currently consuming.
> 3. Is there a better way of doing this?



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Last offset in all partitions

[Todd Palino]

We do this through our monitoring agents by pulling it as a metric from the
LogEndOffset beans. By putting it into our metrics system we get a mapping
of timestamp to offset for every partition with (currently) 60 second
granularity. Useful for offset resets and other tasks.

-Todd

On Wednesday, July 6, 2016, Kristoffer Sjögren <sto...@gmail.com> wrote:

> Hi
>
> Is there a way to get the last offset written by all partitions of a
> topic programmatically using the 0.10.0.0 API?
>
> At the moment I use KafkaConsumer.seekToEnd as seen in this gist[1]
> but maybe there is a better, more efficient, way to do it?
>
> Cheers,
> -Kristoffer
>
> [1] https://gist.github.com/krisskross/a49e462bedb89505e372672cd81129ab
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Delete Message From topic

[Todd Palino]

Well, if you have a log compacted topic, you can issue a tombstone message
(key with a null message) to delete it. Outside of that, what Tom said
applies.

-Todd


On Tue, Jun 14, 2016 at 9:13 PM, Mudit Kumar <mudit.ku...@askme.in> wrote:

> Thanks Tom!
>
>
>
>
> On 6/14/16, 8:55 PM, "Tom Crayford" <tcrayf...@heroku.com> wrote:
>
> >Hi Mudit,
> >
> >Sorry this is not possible. The only deletion Kafka offers is retention or
> >whole topic deletion.
> >
> >Thanks
> >
> >Tom Crayford
> >Heroku Kafka
> >
> >On Tuesday, 14 June 2016, Mudit Kumar <mudit.ku...@askme.in> wrote:
> >
> >> Hey,
> >>
> >> How can I delete particular messages from particular topic?Is that
> >> possible?
> >>
> >> Thanks,
> >> Mudit
> >>
> >>
>
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Question about heterogeneous brokers in a cluster

[Todd Palino]

So as Alex noted, there’s no immediate problem to doing this. Kafka itself
doesn’t know much about the underlying hardware, so it’s not going to care.
At the same time, this means that it has no way natively to know that those
systems have more storage capacity. So they’re not going to automatically
get more partitions.

You have some options here.

   1. You could just ignore it, and treat everything like it’s the smaller
   brokers. This is easy, but you’ll waste your extra storage
   2. You could manually assign more partitions to the larger brokers. This
   requires a little bit of work, but it will more effectively use the
   hardware.

The gotcha with #2 is that you have to make sure you’re not sending too
much network traffic to the larger brokers, and you need to make sure that
you’re not exhausting the CPU as well. And, of course, you’re going to have
to keep an eye on new topics or anything like that to make sure that your
weighted cluster balance is still where you want it to be, and manually fix
it if not.

-Todd


On Fri, Jun 10, 2016 at 5:26 AM, Alex Loddengaard <a...@confluent.io> wrote:

> Hi Kevin,
>
> If you keep the same configs on the new brokers with more storage capacity,
> I don't foresee any issues. Although I haven't tried it myself.
>
> What may introduce headaches is if you have different configuration options
> per broker. Or if you try to assign more partitions to the newer brokers to
> use more of their disk space.
>
> Let's see if others notice anything I'm missing (again, I've never tried
> this before). Hope this helps.
>
> Alex
>
> On Thu, Jun 9, 2016 at 10:27 AM, Kevin A <k4m...@gmail.com> wrote:
>
> > Hi there,
> >
> > I have a couple of Kafka brokers and thinking about adding a few more.
> The
> > new broker machines would have a lot more storage available to them than
> > the existing brokers. Am I setting myself up for operational headaches by
> > deploying a heterogeneous (in terms of storage capacity) cluster?
> >
> > (Asked on IRC but thought I'd try here too.)
> >
> > Thanks!
> > -Kevin
> >
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: JVM Optimizations

[Todd Palino]

; > -XX:InitiatingHeapOccupancyPercent=35
> > > > >
> > > > > Thanks!
> > > > > Ben Osheroff
> > > > > Zendesk.com
> > > > >
> > > > > On Thu, Jun 09, 2016 at 03:52:41PM -0400, Stephen Powis wrote:
> > > > > > NOTE -- GC tuning is outside the realm of my expertise by all
> > means,
> > > so
> > > > > I'm
> > > > > > not sure I'd use our info as any kind of benchmark.
> > > > > >
> > > > > > But in the interest of sharing, we use the following options
> > > > > >
> > > > > > export KAFKA_HEAP_OPTS="-Xmx12G -Xms12G"
> > > > > > >
> > > > > > > export KAFKA_JVM_PERFORMANCE_OPTS="-server
> > -Djava.awt.headless=true
> > > > > > > -XX:MaxPermSize=48M -verbose:gc -Xloggc:/var/log/kafka/gc.log
> > > > > > > -XX:+PrintGCDateStamps -XX:+PrintGCDetails
> > > > > -XX:+PrintTenuringDistribution
> > > > > > > -XX:+PrintGCApplicationStoppedTime -XX:+PrintTLAB
> > > > > -XX:+DisableExplicitGC
> > > > > > > -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10
> > > > > -XX:GCLogFileSize=100M
> > > > > > > -XX:+UseCompressedOops -XX:+AlwaysPreTouch -XX:+UseG1GC
> > > > > > > -XX:MaxGCPauseMillis=20 -XX:+HeapDumpOnOutOfMemoryError
> > > > > > > -XX:HeapDumpPath=/var/log/kafka/heapDump.log"
> > > > > > >
> > > > > >
> > > > > > You can then take your gc.log files and use an analyzer
> tool...I've
> > > > > > attached a link to one of our brokers gclog run thru gceasy.io.
> > > > > >
> > > > > > https://protect-us.mimecast.com/s/wXqqBJuqdZb1Tn
> > > > > >
> > > > > > On Thu, Jun 9, 2016 at 3:39 PM, Lawrence Weikum <
> > lwei...@pandora.com
> > > >
> > > > > wrote:
> > > > > >
> > > > > > > Hi Tom,
> > > > > > >
> > > > > > > Currently we’re using the default settings – no special tuning
> > > > > > > whatsoever.  I think the kafka-run-class.sh has this:
> > > > > > >
> > > > > > >
> > > > > > > # Memory options
> > > > > > > if [ -z "$KAFKA_HEAP_OPTS" ]; then
> > > > > > >   KAFKA_HEAP_OPTS="-Xmx256M"
> > > > > > > fi
> > > > > > >
> > > > > > > # JVM performance options
> > > > > > > if [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; then
> > > > > > >   KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC
> > > > > -XX:MaxGCPauseMillis=20
> > > > > > > -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC
> > > > > > > -Djava.awt.headless=true"
> > > > > > > fi
> > > > > > >
> > > > > > >
> > > > > > > Is this the confluent doc you were referring to?
> > > > > > > https://protect-us.mimecast.com/s/arXXBOspkvORCD
> > > > > > >
> > > > > > > Thanks!
> > > > > > >
> > > > > > > Lawrence Weikum
> > > > > > >
> > > > > > >
> > > > > > > On 6/9/16, 1:32 PM, "Tom Crayford" <tcrayf...@heroku.com>
> wrote:
> > > > > > >
> > > > > > > >Hi Lawrence,
> > > > > > > >
> > > > > > > >What JVM options were you using? There's a few pages in the
> > > > confluent
> > > > > docs
> > > > > > > >on JVM tuning iirc. We simply use the G1 and a 4GB Max heap
> and
> > > > things
> > > > > > > work
> > > > > > > >well (running many thousands of clusters).
> > > > > > > >
> > > > > > > >Thanks
> > > > > > > >Tom Crayford
> > > > > > > >Heroku Kafka
> > > > > > > >
> > > > > > > >On Thursday, 9 June 2016, Lawrence Weikum <
> lwei...@pandora.com>
> > > > > wrote:
> > > > > > > >
> > > > > > > >> Hello all,
> > > > > > > >>
> > > > > > > >> We’ve been running a benchmark test on a Kafka cluster of
> ours
> > > > > running
> > > > > > > >> 0.9.0.1 – slamming it with messages to see when/if things
> > might
> > > > > break.
> > > > > > > >> During our test, we caused two brokers to throw OutOfMemory
> > > errors
> > > > > > > (looks
> > > > > > > >> like from the Heap) even though each machine still has 43%
> of
> > > the
> > > > > total
> > > > > > > >> memory unused.
> > > > > > > >>
> > > > > > > >> I’m curious what JVM optimizations are recommended for Kafka
> > > > > brokers?
> > > > > > > Or
> > > > > > > >> if there aren’t any that are recommended, what are some
> > > > > optimizations
> > > > > > > >> others are using to keep the brokers running smoothly?
> > > > > > > >>
> > > > > > > >> Best,
> > > > > > > >>
> > > > > > > >> Lawrence Weikum
> > > > > > > >>
> > > > > > > >>
> > > > > > >
> > > > > > >
> > > > >
> > > >
> > >
> > >
> > >
> > > --
> > > Dustin Cote
> > > confluent.io
> > >
> >
>
>
>
> --
> Dustin Cote
> confluent.io
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka behind a load balancer

[Todd Palino]

Yep, what Ewen said. We have all of our Kafka clusters behind hardware load
balancers. Producers (and eventually consumers, once we switch to the new
consumer) get configured with those VIPs. It’s better than providing a list
of brokers for the cluster, because we often change the particular brokers
in the cluster.

Note that there is a difference in behavior between the older producer
library and the current one. The old producer would make all metadata
requests through the VIP (the broker list), and then connect directly to
the individual brokers for produce/fetch. The current producer uses the VIP
(bootstrap servers) once at startup to populate the list the list of
brokers for the cluster and then makes metadata and all other requests to
those brokers directly (not through the VIP).

As a side note here, has anyone validated the behavior of the clients when
the bootstrap.servers is a round robin DNS entry?

-Todd


On Sun, Jun 5, 2016 at 6:34 AM, Ewen Cheslack-Postava <e...@confluent.io>
wrote:

> Note, however, that a load balancer can be useful for bootstrapping
> purposes, i.e. use it for the bootstrap.servers setting to have a single
> consistent value for the setting but allow the broker list to change over
> time. From there, as Tom says, it'll start using broker hostnames and
> automatically target the specific brokers it needs to communicate with.
>
> -Ewen
>
> On Fri, Jun 3, 2016 at 5:37 AM, cs user <acldstk...@gmail.com> wrote:
>
> > Hi Tom,
> >
> > That's great, I thought as much, thanks for taking the time to respond,
> > much appreciated!
> >
> > Cheers
> >
> > On Fri, Jun 3, 2016 at 1:18 PM, Tom Crayford <tcrayf...@heroku.com>
> wrote:
> >
> > > Hi,
> > >
> > > Kafka is designed to distribute traffic between brokers itself. It's
> > > naturally distributed and does not need, and indeed will not work
> behind
> > a
> > > load balancer. I'd recommend reading the docs for more, but
> > > http://kafka.apache.org/documentation.html#design_loadbalancing is a
> > good
> > > start.
> > >
> > > Thanks
> > >
> > > Tom Crayford
> > > Heroku Kafka
> > >
> > > On Fri, Jun 3, 2016 at 1:15 PM, cs user <acldstk...@gmail.com> wrote:
> > >
> > > > Hi All,
> > > >
> > > > Does anyone have any experience of using kafka behind a load
> balancer?
> > > >
> > > > Would this work? Are there any reasons why you would not want to do
> it?
> > > >
> > > > Thanks!
> > > >
> > >
> >
>
>
>
> --
> Thanks,
> Ewen
>



-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Change Topic Name

[Todd Palino]

With the caveat that I’ve never tried this before...

I don’t see a reason why this wouldn’t work. There’s no topic information
that’s encoded in the log segments, as far as I’m aware. And there’s no
information about offsets stored in Zookeeper. So in theory, you should be
able to shut down the entire Kafka cluster (this will require a full
shutdown), and copy the log segments from one directory to another.
Obviously, the partition to broker assignments may not be the same, so
you’ll either need to align them with a partition reassignment beforehand,
or make sure you put the right partitions on the right brokers.

-Todd


On Wed, Jun 1, 2016 at 12:02 PM, Vladimir Picka <oss.mli...@gmail.com>
wrote:

> Does creating new topic with new name and the same settings as the
> original one and directly copying files from kafka log directory into the
> new topic folder work? It would be nice if it would. I don't know if there
> is anything in log file format info or Zoo info attached to original topic
> which would prevent it to work.
>
> Petr
>
> -----Original Message-
> From: Todd Palino [mailto:tpal...@gmail.com]
> Sent: 1. června 2016 10:48
> To: users@kafka.apache.org
> Subject: Re: Change Topic Name
>
> There's no way to do that. If you're trying to maintain data, you'll need
> to read all the data from the existing topic and produce it to the new one.
>
> -Todd
>
> On Wednesday, June 1, 2016, Johannes Nachtwey <
> johannes.nachtweyatw...@gmail.com> wrote:
>
> > Hi guys,
> >
> > it´s possible to change an existing topic name?
> >
> > Thanks and best wishes
> > Johannes
> >
>
>
> --
> *Todd Palino*
> Staff Site Reliability Engineer
> Data Infrastructure Streaming
>
>
>
> linkedin.com/in/toddpalino
>
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Change Topic Name

[Todd Palino]

There's no way to do that. If you're trying to maintain data, you'll need
to read all the data from the existing topic and produce it to the new one.

-Todd

On Wednesday, June 1, 2016, Johannes Nachtwey <
johannes.nachtweyatw...@gmail.com> wrote:

> Hi guys,
>
> it´s possible to change an existing topic name?
>
> Thanks and best wishes
> Johannes
>


-- 
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Topics, partitions and keys

[Todd Palino]

Answers are in-line below.

-Todd

On Sun, May 29, 2016 at 3:00 PM, Igor Kravzov <igork.ine...@gmail.com>
wrote:

> Please help me with the subject.
> In Kafka documentations I found the following:
>
> *Kafka only provides a total order over messages within a partition, not
> between different partitions in a topic. Per-partition ordering combined
> with the ability to partition data by key is sufficient for most
> applications. However, if you require a total order over messages this can
> be achieved with a topic that has only one partition, though this will mean
> only one consumer process per consumer group.*
>
> So here are my questions:
> 1. Does it mean if i want to have more than 1 consumer (from the same
> group) reading from the same topic I need to have more than 1 partition?
>

Yes


> 2. Does it mean that I need the same amount of partitions as amount of
> consumers for the same group?
>

If you want all your consumers to be actively consuming, then you need at
least as many partitions as you have consumers. You can have more
partitions than you do consumers, and you can be assured that all
partitions will be consumed. Note that you don’t have to have as many
partitions as you do consumers - you could have some “warm spares” in place
to pick up if others drop out.


> 3. How many consumers can read from one partition?
>

Only one consumer in a given consumer group can read from a single
partition. You can have as many consumer groups reading the same topic as
your system is able to handle (i.e. how much network bandwidth do you have,
for example).


> Also have some questions regarding relationship between keys and partitions
> with regard to API. I only looked at .net APIs (especially one from MS)
>  but looks like the mimic Java API.
> Whhen using a producer to send a message to a topic there is a key
> parameter. But when consumer reads from a topic there is a partition
> number.
>
> 1. How are partitions numbered? Starting from 0 or 1?
>

Starting from zero. A topic with eight partitions will have them numbered
zero through seven.


> 2. What exactly relationship between a key and partition?
> As I understand some function on key will determine a partition. is that
> correct?
>

When using keys, the default partitioner will hash the key and write it to
a partition based on the hash. All messages produced with that same key
will be written to the same partition. This is useful, for example, if you
want to make sure all messages that deal with a particular user are in the
same partition to optimize processing on your consumers.

Note that the hashing and partition assignment will be consistent as long
as the number of partitions remains the same. If you change the number of
partitions for the topic, the assignments of keys to partitions will change.


> 3. If I have 2 partitions in a topic and want some particular messages go
> to one partition and other messages go to another I should use a specific
> key for one specific partition, and the rest for another?
>

For 2 partitions, it is likely, but not guaranteed, that if you produce
messages with 2 different keys they will end up in different partitions. If
you truly want to assure partitioning like that, you will need to use a
custom partitioner. This will allow you to specify, in the producer,
exactly what partition each message gets produced to.


> 4. What if I have 3 partitions and one type of messages to go to one
> particular partition and the rest to other two?
>

Again, this is a case where you need to use a custom partitioner if you
want to get fancy.


> 5. How in general I send messages to a particular partition in order to
> know  for a consumer from where to read?
> Or I better off with multiple topics?
>

Either way is OK. I usually use the guideline that if the messages are of
different types (e.g. one is a page view event, and one is a search event),
they should probably be in different topics named appropriately. This
allows your consumers to know exactly what they are dealing with, and you
never know when you’ll have a consumer that will care about one and not the
other. You want to minimize a consumer reading messages and throwing them
away because they’re not what it wants to be reading.

Doing something like having a topic per-user can be very problematic as the
scale starts to increase. Yes, you can certainly use a wildcard consumer,
but if you’re not doing that you have to maintain some mapping of consumers
to topics. And if you are using a wildcard consumer, you’re going to run
into issues with the number of topics any given group is consuming at some
point. Your system may work fine for 5 topics, but what about when it grows
to 100? 1000? A million?


>
> Thanks in advance.
>



*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka - Networking considerations

[Todd Palino]

One thing to keep in mind is that the 18.5 Gbps number is across the entire
Kafka infrastructure at LinkedIn, not a single cluster. For a single
cluster, I’d say our peak right now is 4.7 Gbps inbound. However, because
Kafka is horizontally scalable, you can build a cluster to support what
you’re looking for.

At the very basic, if we have gigabit network interfaces, 16 Gbps requires
at least 16 brokers. This assumes a replication factor of 1, which is
almost certainly not what you want to do (as there is no protection within
the cluster for a node failure). If you have RF=2, every bit written into
your cluster by producers is going to generate a bit outbound from a broker
for replication, and a bit inbound on the broker it’s being replicated to.
This means that your inbound throughput is halved, and you need at least 32
brokers. This is simplistic - it doesn’t take into account performance, or
retention on disk, both of which you’ll need to think about. You can’t run
the network interfaces saturated all the time, but you can get pretty close
without having performance problems.

You also need to be concerned about your consumers here. If you have a
single consumer, there’s no real complication. It maps directly with your
producers. But I (for example) have 3 consumers (plus the replicas) on
average for any given message. That means I need 4 times as much outbound
bandwidth as I do inbound. If my producers write in 250 Mbps, I’m
saturating that 1 Gbps network link on the other side.

As far as laying out the brokers, this is where things will get a lot more
interesting. Ideally, you want the cluster to have as many failure domains
as possible, so you want to spread it out to lots of racks (to minimize the
impact of a power failure or a single switch failure, assuming a top of
rack switch). Which means you have to take into account the cross-talk
between the brokers for replication. Spine and leaf network topologies work
well here because they don’t involve quite so many cross-connects between
the racks to get big pipes.

Really, you’re going to have to work with your network engineers to figure
out what you have to work with, and plan within that. It’s hard to stand on
the outside and give you a solid plan, because I don’t know what else is
going on on your networks, where your producers are, what your consumers
are doing, or what your performance needs to look like.

-Todd


On Fri, May 6, 2016 at 4:07 AM, Andrew Backhouse <backhouseand...@gmail.com>
wrote:

> Hello,
>
> I trust you are well.
>
> There's a wealth of great articles and presentations relating to the Kafka
> logic and design, thank you for these. But, I'm unable to find information
> relating to the network infrastructure that underpins Kafka. What I am
> trying to understand and put my Network Engineers at ease when it comes to
> sizing our Kafka cluster is as follows;
>
> Looking at an example deployment e.g. LinkedIn and their Kafka deployment,
> it processes 18.5Gbit/sec inbound.
>
> What design considerations were made to;
>
>
>1. validate the network can support such throughput (inbound,
>inter-cluster & outbound)
>2. identify what network infrastructure config/layout can be used or
>avoided to optimise for this level of traffic
>
> To help your answer, we're looking at potentially 16GBit/sec inbound which
> concerns our network team.
>
> If you can please share pointers to existing materials or specific details
> of your deployment, that will be great.
>
>
> Regards,
>
> Andrew
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Reg. Partition Rebalancing

[Todd Palino]

There’s no data moved when a PLE occurs, so that is not the problem here.
The issue is that the controller reelects all partitions at the same time.
This forces all the replica fetchers on the broker to stop and restart (for
the partitions that have changed leaders) at the same time. We have seen,
as well as other users, that this causes all clients to start failing for
several minutes while this problem is ongoing.

There’s an additional problem that the controller appears to start counting
when to perform the PLE (based on the configured waiting period) from “all
the brokers are online”. If you are performing a rolling bounce, this can
conflict seriously with our shutdown check which assures that the cluster
under replicated count is zero before performing a shutdown.

-Todd


On Tue, Mar 29, 2016 at 1:29 PM, James Cheng <jch...@tivo.com> wrote:

>
> > On Mar 29, 2016, at 10:33 AM, Todd Palino <tpal...@gmail.com> wrote:
> >
> > There’s two things that people usually mean when they talk about
> > rebalancing.
> >
> > One is leader reelection, or preferred replica election, which is
> sometimes
> > confusingly referred to as “leader rebalance”. This is when we ask the
> > controller in the cluster to pick the preferred replica for all
> partitions
> > and change which broker is the leader (as appropriate). This is very
> useful
> > when you have to take a broker offline for maintenance, as you can
> quickly
> > make it take over leadership for partitions once it is back online. The
> > controller picks the preferred leader pretty simply: the replica list is
> an
> > array, and the controller picks the first broker listed in the array that
> > is currently in sync as preferred. This means that the PLE is
> > deterministic, and will always give you the same partition leadership
> > (assuming the replicas are in sync).
> >
> > There is an admin CLI command to trigger the preferred replica election
> > manually. There is also a broker configuration
> > “auto.leader.rebalance.enable” which you can set to have the broker
> > automatically perform the PLE when needed. DO NOT USE THIS OPTION. There
> > are serious performance issues when doing so, especially on larger
> > clusters. It needs some development work that has not been fully
> identified
> > yet.
> >
>
> Todd,
>
> What do you mean specifically by "serious performance issues"? I know that
> if you enable "auto.leader.rebalance.enable", then the broker(s) will do
> the reassignment whenever they want, at a time you can't predict. And the
> reassignment can move a lot of data around the cluster, and therefore it is
> undesireable to move so much data around at unpredictable times.
>
> Is that the main performance issue you were talking about, or was there
> something else?
>
> -James
>
>
> > The other thing we mean by rebalance is partition rebalancing, or
> changing
> > which brokers are replicas for a given partition to spread out the
> > partitions according to some algorithm. This is something that you want
> to
> > do when you add a broker to a cluster (or remove it), to redistribute
> load
> > within the cluster. It’s also useful periodically to make sure you have a
> > good spread of load, especially as topics change patterns (like ramping
> new
> > features).
> >
> > While there are admin tools to perform partition reassignments, the
> brokers
> > are not yet that great about rebalancing partitions. There is also
> > currently no automated way of doing this, which is OK because it involves
> > moving a lot of data around. Internally at LinkedIn we have some scripts
> we
> > use for more intelligently balancing partitions to assure even balances
> > based on a number of criteria. I’m hoping to have more to say about this
> > later this week.
> >
> > -Todd
> >
> >
> > On Tue, Mar 29, 2016 at 7:27 AM, Srikanth Chandika <
> linosrika...@gmail.com>
> > wrote:
> >
> >> Hi,
> >>
> >> I am new to kafka I am testing all the options in kafka.
> >> I am confused about the re-balancing?
> >> How and where to configure the re-balancing option?
> >>
> >> Regards,
> >> Srikanth
> >>
> >
> >
> >
> > --
> > *—-*
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
>
>
> 
>
> This email and any attachments may contain confidential and privileged
> material for the sole use of the intended recipient. Any review, copying,
> or distribution of this email (or any attachments) by others is prohibited.
> If you are not the intended recipient, please contact the sender
> immediately and permanently delete this email and any attachments. No
> employee or agent of TiVo Inc. is authorized to conclude any binding
> agreement on behalf of TiVo Inc. by email. Binding agreements with TiVo
> Inc. may only be made by a signed written agreement.
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Reg. Partition Rebalancing

[Todd Palino]

There’s two things that people usually mean when they talk about
rebalancing.

One is leader reelection, or preferred replica election, which is sometimes
confusingly referred to as “leader rebalance”. This is when we ask the
controller in the cluster to pick the preferred replica for all partitions
and change which broker is the leader (as appropriate). This is very useful
when you have to take a broker offline for maintenance, as you can quickly
make it take over leadership for partitions once it is back online. The
controller picks the preferred leader pretty simply: the replica list is an
array, and the controller picks the first broker listed in the array that
is currently in sync as preferred. This means that the PLE is
deterministic, and will always give you the same partition leadership
(assuming the replicas are in sync).

There is an admin CLI command to trigger the preferred replica election
manually. There is also a broker configuration
“auto.leader.rebalance.enable” which you can set to have the broker
automatically perform the PLE when needed. DO NOT USE THIS OPTION. There
are serious performance issues when doing so, especially on larger
clusters. It needs some development work that has not been fully identified
yet.

The other thing we mean by rebalance is partition rebalancing, or changing
which brokers are replicas for a given partition to spread out the
partitions according to some algorithm. This is something that you want to
do when you add a broker to a cluster (or remove it), to redistribute load
within the cluster. It’s also useful periodically to make sure you have a
good spread of load, especially as topics change patterns (like ramping new
features).

While there are admin tools to perform partition reassignments, the brokers
are not yet that great about rebalancing partitions. There is also
currently no automated way of doing this, which is OK because it involves
moving a lot of data around. Internally at LinkedIn we have some scripts we
use for more intelligently balancing partitions to assure even balances
based on a number of criteria. I’m hoping to have more to say about this
later this week.

-Todd


On Tue, Mar 29, 2016 at 7:27 AM, Srikanth Chandika <linosrika...@gmail.com>
wrote:

> Hi,
>
> I am new to kafka I am testing all the options in kafka.
> I am confused about the re-balancing?
> How and where to configure the re-balancing option?
>
> Regards,
> Srikanth
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: [kafka] decreasing Lag

[Todd Palino]

So it looks like you need more logstash consumers, but you’ll want to look
at the consumers you have and make sure they are working well and they’re
not getting bogged down somewhere else, which is causing them to consume
slower. Assuming they’re working fine, you can add 4 more.

If that doesn’t drop it down, you can then look at increasing the number of
partitions and increasing the number of logstash consumers further. While
you may get some benefit from increasing partitions without increasing the
consumer count, you’ll most likely have to do both.

-Todd


On Mon, Mar 7, 2016 at 8:46 AM, Tim Desrochers <tgdesroch...@gmail.com>
wrote:

> I am new to Kafka so please excuse me if this is a very basic question.
>
> I have a cluster set up with 3 zookeepers and 9 brokers.  I have network
> security logs flowing into the kafka cluster.  I am using logstash to read
> them from the cluster and ingest them into an elasticsearch cluster.
>
> My current settings are mostly default.  I created a topic with 8
> partitions.  I have 4 logstash consumers reading that topic and feeding my
> ES cluster.  My problem is I can't keep up with real time at the moment.  I
> am constantly falling behind and logs are building on my kafka cluster.
>
> When I run:
> $ /opt/kafka/bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker
> --group logstash --zookeeper localhost:2181 --topic bro-logs
>
> I get the following:
> logstashbro-logs   0   2593739429935485
>3998091 logstash_OP-01-VM-553-1457301346564-d14fd84a-0
> logstashbro-logs   1   2592959429935506
>4005912 logstash_OP-01-VM-553-1457301346564-d14fd84a-0
> logstashbro-logs   2   2671072829935519
>3224791 logstash_OP-01-VM-554-1457356976268-fa8c24b9-0
> logstashbro-logs   3   3887940 6372075
> 2484135 logstash_OP-01-VM-554-1457356976268-fa8c24b9-0
> logstashbro-logs   4   3978342 6372074
> 2393732 logstash_OP-01-VM-555-1457368235387-c6b8bd1f-0
> logstashbro-logs   5   3984965 6372075
> 2387110 logstash_OP-01-VM-555-1457368235387-c6b8bd1f-0
> logstashbro-logs   6   4017715 6372076
> 2354361 logstash_OP-01-VM-556-1457368464998-8edb13df-0
> logstashbro-logs   7   4022484 6372074
> 2349590 logstash_OP-01-VM-556-1457368464998-8edb13df-0
>
> from what I understand the Lag column is telling me that there are a hole
> bunch of logs waiting in the cluster to be processed.
>
> So my question is, should I spin up more logstash consumers to read from
> the kafka cluster and feed the ES cluster?  Should I increase or decrease
> partitions?  What can be done to increase the amount of logs being read
> from the cluster and ingested into Elastisearch?
>
> Like I said, very new to kafka.
>
> Thanks for the help
> Tim
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Apache Kafka Case Studies

[Todd Palino]

To follow up on the blog post of mine that Jens pasted, I also have a
number of talk presentations up at http://www.slideshare.net/toddpalino

If you download the actual presentation files, you’ll be able to access the
slide notes in most of those that have a lot more information.

On Wed, Feb 3, 2016 at 7:47 AM, Jens Rantil <jens.ran...@tink.se> wrote:

> Hi Joe,
>
> This might be interesting:
> https://engineering.linkedin.com/kafka/running-kafka-scale
>
> Cheers,
> Jens
>
> On Wed, Feb 3, 2016 at 4:15 PM, Joe San <codeintheo...@gmail.com> wrote:
>
> > Dear Kafka users,
> >
> > I'm looking for some case studies around using Kafka on big projects.
> > Specifically, I'm looking for some architectural insights into how I
> could
> > orchestrate my data pipeline using Kafka on an enterprise system.
> >
> > Some pointers on some architectural best practices, slides on how some
> > organisation X used Apache Kafka in their landscape would be ideal.
> >
> > Any suggestions?
> >
> > Thanks and Regards,
> > Joe
> >
>
>
>
> --
> Jens Rantil
> Backend engineer
> Tink AB
>
> Email: jens.ran...@tink.se
> Phone: +46 708 84 18 32
> Web: www.tink.se
>
> Facebook <https://www.facebook.com/#!/tink.se> Linkedin
> <
> http://www.linkedin.com/company/2735919?trk=vsrp_companies_res_photo=VSRPsearchId%3A1057023381369207406670%2CVSRPtargetId%3A2735919%2CVSRPcmpt%3Aprimary
> >
>  Twitter <https://twitter.com/tink>
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Kafka + ZooKeeper on the same hardware?

[Todd Palino]

I’d say it depends on load and usage. It can definitely be done, and we’ve
done it here in places, though we don’t anymore. Part of the luxury of
being able to get the hardware we want. In general, it’s probably easier to
do with 0.9 and Kafka-committed offsets, because the consumers don’t need
to talk to ZK as much. It’s probably even even easier with the new
consumer, but I can’t speak to that at all.

One of the gotchas is that ZK really should have its transaction log on an
isolate device so that sequential writes do not require seeks. This could
be a separate disk or an SSD drive. An example of a really bad place to put
it would be on the same device as your Kafka log segments :) Depending on
your load, it may not be critical to use a separate device.

As Gwen noted, it all comes down to load. Your availability will be fine,
you just need to figure out if the services can share the load.

-Todd


On Thu, Jan 14, 2016 at 9:25 AM, Gwen Shapira <g...@confluent.io> wrote:

> It depends on load :)
> As long as there is no contention, you are fine.
>
> On Thu, Jan 14, 2016 at 6:06 AM, Erik Forsberg <forsb...@opera.com> wrote:
>
> > Hi!
> >
> > Pondering how to configure Kafka clusters and avoid having too many
> > machines to manage.. Would it be recommended to run say a 3 node kafka
> > cluster where you also run your 3 node zookeeper cluster on the same
> > machines?
> >
> > I guess the answer is that "it depends on load", but would be interested
> > in any opinions on this anyway.
> >
> > Thanks!
> > \EF
> >
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: how to reset kafka offset in zookeeper

[Todd Palino]

If what you want to do is reset to smallest, all you need to do is stop the
consumer, delete the group from Zookeeper, and restart the consumer. It
will automatically create the group again.

You only need to export the offsets first if you later need to reset to
where you were in the partitions.

-Todd

On Saturday, December 19, 2015, Akhilesh Pathodia <
pathodia.akhil...@gmail.com> wrote:

> What is the process for deleting the consumer group from zookeeper? Should
> I export offset, delete and then import?
>
> Thanks,
> Akhilesh
>
> On Fri, Dec 18, 2015 at 11:32 PM, Todd Palino <tpal...@gmail.com
> <javascript:;>> wrote:
>
> > Yes, that’s right. It’s just work for no real gain :)
> >
> > -Todd
> >
> > On Fri, Dec 18, 2015 at 9:38 AM, Marko Bonaći <marko.bon...@sematext.com
> <javascript:;>>
> > wrote:
> >
> > > Hmm, I guess you're right Tod :)
> > > Just to confirm, you meant that, while you're changing the exported
> file
> > it
> > > might happen that one of the segment files becomes eligible for cleanup
> > by
> > > retention, which would then make the imported offsets out of range?
> > >
> > > Marko Bonaći
> > > Monitoring | Alerting | Anomaly Detection | Centralized Log Management
> > > Solr & Elasticsearch Support
> > > Sematext <http://sematext.com/> | Contact
> > > <http://sematext.com/about/contact.html>
> > >
> > > On Fri, Dec 18, 2015 at 6:29 PM, Todd Palino <tpal...@gmail.com
> <javascript:;>> wrote:
> > >
> > > > That works if you want to set to an arbitrary offset, Marko. However
> in
> > > the
> > > > case the OP described, wanting to reset to smallest, it is better to
> > just
> > > > delete the consumer group and start the consumer with
> auto.offset.reset
> > > set
> > > > to smallest. The reason is that while you can pull the current
> smallest
> > > > offsets from the brokers and set them in Zookeeper for the consumer,
> by
> > > the
> > > > time you do that the smallest offset is likely no longer valid. This
> > > means
> > > > you’re going to resort to the offset reset logic anyways.
> > > >
> > > > -Todd
> > > >
> > > >
> > > > On Fri, Dec 18, 2015 at 7:10 AM, Marko Bonaći <
> > marko.bon...@sematext.com <javascript:;>
> > > >
> > > > wrote:
> > > >
> > > > > You can also do this:
> > > > > 1. stop consumers
> > > > > 2. export offsets from ZK
> > > > > 3. make changes to the exported file
> > > > > 4. import offsets to ZK
> > > > > 5. start consumers
> > > > >
> > > > > e.g.
> > > > > bin/kafka-run-class.sh kafka.tools.ExportZkOffsets --group
> group-name
> > > > > --output-file /tmp/zk-offsets --zkconnect localhost:2181
> > > > > bin/kafka-run-class.sh kafka.tools.ImportZkOffsets --input-file
> > > > > /tmp/zk-offsets --zkconnect localhost:2181
> > > > >
> > > > > Marko Bonaći
> > > > > Monitoring | Alerting | Anomaly Detection | Centralized Log
> > Management
> > > > > Solr & Elasticsearch Support
> > > > > Sematext <http://sematext.com/> | Contact
> > > > > <http://sematext.com/about/contact.html>
> > > > >
> > > > > On Fri, Dec 18, 2015 at 4:06 PM, Jens Rantil <jens.ran...@tink.se
> <javascript:;>>
> > > > wrote:
> > > > >
> > > > > > Hi,
> > > > > >
> > > > > > I noticed that a consumer in the new consumer API supports
> setting
> > > the
> > > > > > offset for a partition to beginning. I assume doing so also would
> > > > update
> > > > > > the offset in Zookeeper eventually.
> > > > > >
> > > > > > Cheers,
> > > > > > Jens
> > > > > >
> > > > > > On Friday, December 18, 2015, Akhilesh Pathodia <
> > > > > > pathodia.akhil...@gmail.com <javascript:;>>
> > > > > > wrote:
> > > > > >
> > > > > > > Hi,
> > > > > > >
> > > > > > > I want to reset the kafka offset in zookeeper so that the
> > consumer
> > > > will
> > > > > > > start reading m

Re: how to reset kafka offset in zookeeper

[Todd Palino]

There’s no simple command. You’ll need to use either zookeeper-shell.sh or
zkCli.sh or something similar that lets you explore and edit Zookeeper and
do a recursive delete on the group name in the consumers tree. I’m not sure
how Cloudera’s interface differs, however, or if they provide a separate
tool for deleting a consumer group.

-Todd


On Sat, Dec 19, 2015 at 11:34 AM, Akhilesh Pathodia <
pathodia.akhil...@gmail.com> wrote:

> What is the command  to delete  group from zookeeper? I dont find
> /consumer/ directory? I am using cloudera, is there any place on cloudera
> manager where I can delete the group?
>
> Thanks
>
> On Sat, Dec 19, 2015 at 11:47 PM, Todd Palino <tpal...@gmail.com> wrote:
>
> > If what you want to do is reset to smallest, all you need to do is stop
> the
> > consumer, delete the group from Zookeeper, and restart the consumer. It
> > will automatically create the group again.
> >
> > You only need to export the offsets first if you later need to reset to
> > where you were in the partitions.
> >
> > -Todd
> >
> > On Saturday, December 19, 2015, Akhilesh Pathodia <
> > pathodia.akhil...@gmail.com> wrote:
> >
> > > What is the process for deleting the consumer group from zookeeper?
> > Should
> > > I export offset, delete and then import?
> > >
> > > Thanks,
> > > Akhilesh
> > >
> > > On Fri, Dec 18, 2015 at 11:32 PM, Todd Palino <tpal...@gmail.com
> > > <javascript:;>> wrote:
> > >
> > > > Yes, that’s right. It’s just work for no real gain :)
> > > >
> > > > -Todd
> > > >
> > > > On Fri, Dec 18, 2015 at 9:38 AM, Marko Bonaći <
> > marko.bon...@sematext.com
> > > <javascript:;>>
> > > > wrote:
> > > >
> > > > > Hmm, I guess you're right Tod :)
> > > > > Just to confirm, you meant that, while you're changing the exported
> > > file
> > > > it
> > > > > might happen that one of the segment files becomes eligible for
> > cleanup
> > > > by
> > > > > retention, which would then make the imported offsets out of range?
> > > > >
> > > > > Marko Bonaći
> > > > > Monitoring | Alerting | Anomaly Detection | Centralized Log
> > Management
> > > > > Solr & Elasticsearch Support
> > > > > Sematext <http://sematext.com/> | Contact
> > > > > <http://sematext.com/about/contact.html>
> > > > >
> > > > > On Fri, Dec 18, 2015 at 6:29 PM, Todd Palino <tpal...@gmail.com
> > > <javascript:;>> wrote:
> > > > >
> > > > > > That works if you want to set to an arbitrary offset, Marko.
> > However
> > > in
> > > > > the
> > > > > > case the OP described, wanting to reset to smallest, it is better
> > to
> > > > just
> > > > > > delete the consumer group and start the consumer with
> > > auto.offset.reset
> > > > > set
> > > > > > to smallest. The reason is that while you can pull the current
> > > smallest
> > > > > > offsets from the brokers and set them in Zookeeper for the
> > consumer,
> > > by
> > > > > the
> > > > > > time you do that the smallest offset is likely no longer valid.
> > This
> > > > > means
> > > > > > you’re going to resort to the offset reset logic anyways.
> > > > > >
> > > > > > -Todd
> > > > > >
> > > > > >
> > > > > > On Fri, Dec 18, 2015 at 7:10 AM, Marko Bonaći <
> > > > marko.bon...@sematext.com <javascript:;>
> > > > > >
> > > > > > wrote:
> > > > > >
> > > > > > > You can also do this:
> > > > > > > 1. stop consumers
> > > > > > > 2. export offsets from ZK
> > > > > > > 3. make changes to the exported file
> > > > > > > 4. import offsets to ZK
> > > > > > > 5. start consumers
> > > > > > >
> > > > > > > e.g.
> > > > > > > bin/kafka-run-class.sh kafka.tools.ExportZkOffsets --group
> > > group-name
> > > > > > > --output-file /tmp/zk-offsets --zkconnect localhost:2181
> > > > > > > bin/kafka-run-class.sh kafka.tools.ImportZkOffsets --i

Re: how to reset kafka offset in zookeeper

[Todd Palino]

The way to reset to smallest is to stop the consumer, delete the consumer
group from Zookeeper, and then restart with the property set to smallest.
Once your consumer has recreated the group and committed offsets, you can
change the auto.offset.reset property back to largest (if that is your
preference).

-Todd

On Friday, December 18, 2015, Akhilesh Pathodia <pathodia.akhil...@gmail.com>
wrote:

> Hi,
>
> I want to reset the kafka offset in zookeeper so that the consumer will
> start reading messages from first offset. I am using flume as a consumer to
> kafka. I have set the kafka property kafka.auto.offset.reset to "smallest",
> but it does not reset the offset in zookeeper and that's why flume will not
> read messages from first offset.
>
> Is there any way to reset kafka offset in zookeeper?
>
> Thanks,
> Akhilesh
>


-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: how to reset kafka offset in zookeeper

[Todd Palino]

That works if you want to set to an arbitrary offset, Marko. However in the
case the OP described, wanting to reset to smallest, it is better to just
delete the consumer group and start the consumer with auto.offset.reset set
to smallest. The reason is that while you can pull the current smallest
offsets from the brokers and set them in Zookeeper for the consumer, by the
time you do that the smallest offset is likely no longer valid. This means
you’re going to resort to the offset reset logic anyways.

-Todd


On Fri, Dec 18, 2015 at 7:10 AM, Marko Bonaći <marko.bon...@sematext.com>
wrote:

> You can also do this:
> 1. stop consumers
> 2. export offsets from ZK
> 3. make changes to the exported file
> 4. import offsets to ZK
> 5. start consumers
>
> e.g.
> bin/kafka-run-class.sh kafka.tools.ExportZkOffsets --group group-name
> --output-file /tmp/zk-offsets --zkconnect localhost:2181
> bin/kafka-run-class.sh kafka.tools.ImportZkOffsets --input-file
> /tmp/zk-offsets --zkconnect localhost:2181
>
> Marko Bonaći
> Monitoring | Alerting | Anomaly Detection | Centralized Log Management
> Solr & Elasticsearch Support
> Sematext <http://sematext.com/> | Contact
> <http://sematext.com/about/contact.html>
>
> On Fri, Dec 18, 2015 at 4:06 PM, Jens Rantil <jens.ran...@tink.se> wrote:
>
> > Hi,
> >
> > I noticed that a consumer in the new consumer API supports setting the
> > offset for a partition to beginning. I assume doing so also would update
> > the offset in Zookeeper eventually.
> >
> > Cheers,
> > Jens
> >
> > On Friday, December 18, 2015, Akhilesh Pathodia <
> > pathodia.akhil...@gmail.com>
> > wrote:
> >
> > > Hi,
> > >
> > > I want to reset the kafka offset in zookeeper so that the consumer will
> > > start reading messages from first offset. I am using flume as a
> consumer
> > to
> > > kafka. I have set the kafka property kafka.auto.offset.reset to
> > "smallest",
> > > but it does not reset the offset in zookeeper and that's why flume will
> > not
> > > read messages from first offset.
> > >
> > > Is there any way to reset kafka offset in zookeeper?
> > >
> > > Thanks,
> > > Akhilesh
> > >
> >
> >
> > --
> > Jens Rantil
> > Backend engineer
> > Tink AB
> >
> > Email: jens.ran...@tink.se
> > Phone: +46 708 84 18 32
> > Web: www.tink.se
> >
> > Facebook <https://www.facebook.com/#!/tink.se> Linkedin
> > <
> >
> http://www.linkedin.com/company/2735919?trk=vsrp_companies_res_photo=VSRPsearchId%3A1057023381369207406670%2CVSRPtargetId%3A2735919%2CVSRPcmpt%3Aprimary
> > >
> >  Twitter <https://twitter.com/tink>
> >
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: how to reset kafka offset in zookeeper

[Todd Palino]

Yes, that’s right. It’s just work for no real gain :)

-Todd

On Fri, Dec 18, 2015 at 9:38 AM, Marko Bonaći <marko.bon...@sematext.com>
wrote:

> Hmm, I guess you're right Tod :)
> Just to confirm, you meant that, while you're changing the exported file it
> might happen that one of the segment files becomes eligible for cleanup by
> retention, which would then make the imported offsets out of range?
>
> Marko Bonaći
> Monitoring | Alerting | Anomaly Detection | Centralized Log Management
> Solr & Elasticsearch Support
> Sematext <http://sematext.com/> | Contact
> <http://sematext.com/about/contact.html>
>
> On Fri, Dec 18, 2015 at 6:29 PM, Todd Palino <tpal...@gmail.com> wrote:
>
> > That works if you want to set to an arbitrary offset, Marko. However in
> the
> > case the OP described, wanting to reset to smallest, it is better to just
> > delete the consumer group and start the consumer with auto.offset.reset
> set
> > to smallest. The reason is that while you can pull the current smallest
> > offsets from the brokers and set them in Zookeeper for the consumer, by
> the
> > time you do that the smallest offset is likely no longer valid. This
> means
> > you’re going to resort to the offset reset logic anyways.
> >
> > -Todd
> >
> >
> > On Fri, Dec 18, 2015 at 7:10 AM, Marko Bonaći <marko.bon...@sematext.com
> >
> > wrote:
> >
> > > You can also do this:
> > > 1. stop consumers
> > > 2. export offsets from ZK
> > > 3. make changes to the exported file
> > > 4. import offsets to ZK
> > > 5. start consumers
> > >
> > > e.g.
> > > bin/kafka-run-class.sh kafka.tools.ExportZkOffsets --group group-name
> > > --output-file /tmp/zk-offsets --zkconnect localhost:2181
> > > bin/kafka-run-class.sh kafka.tools.ImportZkOffsets --input-file
> > > /tmp/zk-offsets --zkconnect localhost:2181
> > >
> > > Marko Bonaći
> > > Monitoring | Alerting | Anomaly Detection | Centralized Log Management
> > > Solr & Elasticsearch Support
> > > Sematext <http://sematext.com/> | Contact
> > > <http://sematext.com/about/contact.html>
> > >
> > > On Fri, Dec 18, 2015 at 4:06 PM, Jens Rantil <jens.ran...@tink.se>
> > wrote:
> > >
> > > > Hi,
> > > >
> > > > I noticed that a consumer in the new consumer API supports setting
> the
> > > > offset for a partition to beginning. I assume doing so also would
> > update
> > > > the offset in Zookeeper eventually.
> > > >
> > > > Cheers,
> > > > Jens
> > > >
> > > > On Friday, December 18, 2015, Akhilesh Pathodia <
> > > > pathodia.akhil...@gmail.com>
> > > > wrote:
> > > >
> > > > > Hi,
> > > > >
> > > > > I want to reset the kafka offset in zookeeper so that the consumer
> > will
> > > > > start reading messages from first offset. I am using flume as a
> > > consumer
> > > > to
> > > > > kafka. I have set the kafka property kafka.auto.offset.reset to
> > > > "smallest",
> > > > > but it does not reset the offset in zookeeper and that's why flume
> > will
> > > > not
> > > > > read messages from first offset.
> > > > >
> > > > > Is there any way to reset kafka offset in zookeeper?
> > > > >
> > > > > Thanks,
> > > > > Akhilesh
> > > > >
> > > >
> > > >
> > > > --
> > > > Jens Rantil
> > > > Backend engineer
> > > > Tink AB
> > > >
> > > > Email: jens.ran...@tink.se
> > > > Phone: +46 708 84 18 32
> > > > Web: www.tink.se
> > > >
> > > > Facebook <https://www.facebook.com/#!/tink.se> Linkedin
> > > > <
> > > >
> > >
> >
> http://www.linkedin.com/company/2735919?trk=vsrp_companies_res_photo=VSRPsearchId%3A1057023381369207406670%2CVSRPtargetId%3A2735919%2CVSRPcmpt%3Aprimary
> > > > >
> > > >  Twitter <https://twitter.com/tink>
> > > >
> > >
> >
> >
> >
> > --
> > *—-*
> > *Todd Palino*
> > Staff Site Reliability Engineer
> > Data Infrastructure Streaming
> >
> >
> >
> > linkedin.com/in/toddpalino
> >
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Number of partitions and disks in a topic

[Todd Palino]

Getting the partitioning right now is only important if your messages are
keyed. If they’re not, stop reading, start with a fairly low number of
partitions, and expand as needed.

1000 partitions per topic is generally not normal. It’s not really a
problem, but you want to size topics appropriately. Every partition
represents open file handles and overhead on the cluster controller. But if
you’re working with keyed messages, size for your eventual data size. We
use a general guideline of keeping partitions on disk under 25 GB (for 4
days of retention - so ~6 GB of compressed messages per day). We find this
gives us a good spread of data in the cluster, and represents a reasonable
amount of network throughput per partition, so it allows us to scale
easily. It also makes for fewer issues with replication within the cluster,
and mirroring to other clusters.

Outside of a guideline like that, partition based on how you want to spread
out your keys. We have a user who wanted 720 partitions for a given topic
because it has a large number of factors, which allows them to run a
variety of counts of consumers and have balanced load.

As far as multiple disks goes, yes, Kafka can make use of multiple log
dirs. However, there are caveats. It’s fairly naive about how it assigns
partitions to disks, and partitions are assigned by the controller to a
broker with no knowledge of the disks underneath. The broker then makes the
assignment to a single disk. In addition, there’s no tool for moving
partitions from one mount point to another without shutting down the broker
and doing it manually.

-Todd

On Tue, Dec 1, 2015 at 4:31 AM, Guillermo Ortiz <konstt2...@gmail.com>
wrote:

> Hello,
>
> I want to size the kafka cluster with just one topic and I'm going to
> process the data with Spark and others applications.
>
> If I have six hard drives per node, is it kafka smart enough to deal with
> them? I guess that the memory should be very important in this point and
> all data is cached in memory. Is it possible to config kafka to use many
> directories as HDFS, each one with a different disk?
>
> I'm not sure about the number of partitions either. I have read
>
> http://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster/
> and they talk about number of partitions much higher that I had thought. Is
> it normal to have a topic with 1000 partitions? I was thinking about about
> two/four partitions per node. is it wrong my thought?
>
> As I'm going to process data with Spark, I could have numberPartitions
> equals numberExecutors in Spark as max, always thinking in the future and
> sizing higher than that.
>



-- 
*—-*
*Todd Palino*
Staff Site Reliability Engineer
Data Infrastructure Streaming



linkedin.com/in/toddpalino

Re: Help on understanding kafka-topics.sh output

[Todd Palino]

Replicas and Isr are both a comma separated list of broker IDs. So in this
output, I am seeing that you have two Kafka brokers with IDs 1 and 2. You
have a topic, capture, with 16 partitions at replication factor 1 (1
replica per partition). The broker with ID 2 is not online, which is why it
shows in the Replica list for some partitions (meaning that it is assigned
to be a replica), but not in the Isr list (which would indicate that it is
currently in-sync).

The Leader field is the broker ID which is currently the leader for that
partition. For the partitions that are assigned to broker 1, you see that
broker 1 is the leader. For the partitions that are assigned to broker 2,
the leader is listed as -1, which indicates that there is no available
leader. These partitions are considered offline and cannot be produced to
or consumed from. When broker 2 comes back online, the controller will
perform an unclean leader election and select broker 2 (the only replica
available) as the leader for those partitions.

-Todd


On Sun, Nov 22, 2015 at 11:39 AM, Jan Algermissen <
algermissen1...@icloud.com> wrote:

> Hi,
>
> I have a topic with 16 partitions that shows the following output for
>
> kafka-topics.sh --zookeeper x:2181 --topic capture --describe
>
> Can anyone explain to me what the difference in replicas means and what
> Leader of -1 means?
>
> In the logs of my producer I see that no messages seem to be sent to the
> partitions with '-1' and th eproducer buffer becomes exhausted afetr a
> while (maybe that is related?)
>
> Jan
>
> Topic:capture   PartitionCount:16   ReplicationFactor:1
>  Configs:
>
> Topic: capture  Partition: 0Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 1Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 2Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 3Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 4Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 5Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 6Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 7Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 8Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 9Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 10   Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 11   Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 12   Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 13   Leader: 1   Replicas: 1
>  Isr: 1
> Topic: capture  Partition: 14   Leader: -1  Replicas: 2
>  Isr:
> Topic: capture  Partition: 15   Leader: 1   Replicas: 1
>  Isr: 1

Re: Help on understanding kafka-topics.sh output

[Todd Palino]

Hopefully one of the developers can jump in here. I believe there is a
future you can use to get the errors back from the producer. In addition,
you should check the following configs on the producer:

request.required.acks - this controls whether or not your producer is going
to wait for an acknowledgement from the broker, and how many brokers it
waits for
request.timeout.ms - how long the producer waits to satisfy the acks
setting before marking the request failed
retry.backoff.ms - how long the producer waits between retries
message.send.max.retries - the maximum number of retries the producer will
attempt a failed request

-Todd


On Sun, Nov 22, 2015 at 12:31 PM, Jan Algermissen <
algermissen1...@icloud.com> wrote:

> Hi Todd,
>
> yes, correct - thanks.
>
> However, what I am not getting is that the KafkaProducer (see my other
> mail from today) silently accepts the messages and fills them up in the
> buffer until it is exhausted instead of saying that the broker is not
> reachable.
>
> IOW, it seems from an application perspective I am unable to detect that
> messages are not being sent out. Is this normal behavior and I am simply
> doing something wrong or could it be a producer bug?
>
> Jan
>
> Config and code again:
>
> ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
> ProducerConfig.RETRIES_CONFIG -> "0",
> ProducerConfig.ACKS_CONFIG -> "1",
> ProducerConfig.COMPRESSION_TYPE_CONFIG -> "none",
> ProducerConfig.TIMEOUT_CONFIG -> new Integer(3),
> // ProducerConfig.BATCH_SIZE_CONFIG -> new Integer(16384),
> ProducerConfig.BATCH_SIZE_CONFIG -> new Integer(10),
> ProducerConfig.BUFFER_MEMORY_CONFIG -> new Integer(66554432),
> ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG -> new java.lang.Boolean(false),
> ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG ->
> "org.apache.kafka.common.serialization.StringSerializer",
> ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG ->
> "org.apache.kafka.common.serialization.StringSerializer"
>
>
> 
>
> kafkaProducer.send(new ProducerRecord[String,String](topic, key, data),new
> Callback {
>  def onCompletion(recordMetadata: RecordMetadata, e: Exception):Unit = {
>if(e != null) {
>      logger.error(s"Could not send $data",e)
>}
>logger.info("The offset of the record we just sent is: " +
> recordMetadata.offset())
>()
>  }
>
> })
>
>
> > On 22 Nov 2015, at 20:49, Todd Palino <tpal...@gmail.com> wrote:
> >
> > Replicas and Isr are both a comma separated list of broker IDs. So in
> this
> > output, I am seeing that you have two Kafka brokers with IDs 1 and 2. You
> > have a topic, capture, with 16 partitions at replication factor 1 (1
> > replica per partition). The broker with ID 2 is not online, which is why
> it
> > shows in the Replica list for some partitions (meaning that it is
> assigned
> > to be a replica), but not in the Isr list (which would indicate that it
> is
> > currently in-sync).
> >
> > The Leader field is the broker ID which is currently the leader for that
> > partition. For the partitions that are assigned to broker 1, you see that
> > broker 1 is the leader. For the partitions that are assigned to broker 2,
> > the leader is listed as -1, which indicates that there is no available
> > leader. These partitions are considered offline and cannot be produced to
> > or consumed from. When broker 2 comes back online, the controller will
> > perform an unclean leader election and select broker 2 (the only replica
> > available) as the leader for those partitions.
> >
> > -Todd
> >
> >
> > On Sun, Nov 22, 2015 at 11:39 AM, Jan Algermissen <
> > algermissen1...@icloud.com> wrote:
> >
> >> Hi,
> >>
> >> I have a topic with 16 partitions that shows the following output for
> >>
> >> kafka-topics.sh --zookeeper x:2181 --topic capture --describe
> >>
> >> Can anyone explain to me what the difference in replicas means and what
> >> Leader of -1 means?
> >>
> >> In the logs of my producer I see that no messages seem to be sent to the
> >> partitions with '-1' and th eproducer buffer becomes exhausted afetr a
> >> while (maybe that is related?)
> >>
> >> Jan
> >>
> >>Topic:capture   PartitionCount:16   ReplicationFactor:1
> >> Configs:
> >>
> >>Topic: capture  Partition: 0Leader: 1   Replicas: 1
> >> Isr: 1
> >>Topic: capture  Partition: 1Leader: 1   Replicas: 1
> >> Isr: 1
> >

Re: Change kafka broker ids dynamically

[Todd Palino]

I’m not quite sure why you would need to do this - the broker IDs are not
significant outside of the internal metadata. But this is what you would
have to do for each move (assuming you are running with at least
replication factor 2):

1) Shut down the broker
2) Clear its partition data
3) Reconfigure the broker to the new ID
4) Restart the broker
5) Issue a partition reassignment to reassign all of broker 1’s partitions
to broker 0
6) Wait for the broker to replicate all it’s partitions from other members
of the cluster

That’s a lot of moving data around, just to renumber. You can’t just issue
the reassignment while the broker is down, and not delete the partitions,
because the ID number 0 is unknown, so the reassignment will fail (the
broker is not online). If you wanted to shut the entire cluster down you
could, in theory, walk through the Zookeeper tree manually changing all the
replica information. That assumes you can shut the whole cluster down for a
while.

-Todd



On Fri, Nov 6, 2015 at 1:23 PM, Arathi Maddula 
wrote:

> Hi,
> Is it possible to change the broker.ids property for a node belonging to a
> Kafka cluster? For example, currently if I  have brokers with ids 1,2,3. If
> I want to stop broker 1,  can I change broker.id to 0 (with current id =
> 1) in server.properties and meta.properties files and then restart broker
> 1. Can I repeat this for brokers 2 and 3 as well?
>
> Thanks,
> Arathi
>
>

Re: Load balancer for Kafka brokers

[Todd Palino]

We use loadbalancers for our producer configurations, but what you need to
keep in mind is that that connection is only used for metadata requests.
The producer queries the loadbalancer IP for metadata for the topic, then
disconnects and reconnects directly to the Kafka brokers for producing
messages. With the older producer lib, it periodically reconnects to the
loadbalancer to refresh metadata. With the newer producer lib, it actually
caches information about all the brokers locally and queries them directly
for metadata refreshes moving forwards (and therefore does not use the
loadbalancer again).

In your situation, it sounds like you want to put all the individual broker
connections through the GSLB as well. In order to do this, you would have
to:

- have an individual GSLB configuration for each broker, where that config
has an active/passive setup with 1 broker from each DC (Not too bad)
- configure the announced hostnames for each broker to be the same in the
active and passive DC (A little tricky)
- maintain the exact same partition to broker mapping, including
leadership, in each DC (Virtually impossible)

In short, I don’t think this is a reasonable thing to do. You’re not going
to be able to assure the exact partition mapping, especially not in the
face of Zookeeper timeouts or hardware failures that will cause partition
leadership to move around. This will result in partitions becoming
unavailable as soon as one of the clusters shifts just a little bit.

A better way to approach this is probably to set up a front-end service,
such as a REST endpoint for Kafka, which receives produce requests and
publishes them to the local Kafka cluster. Then you can put that endpoint
behind the GSLB, and you do not have to worry about the makeup of the Kafka
clusters themselves. Your producers would all send their messages through
the GSLB to that endpoint, rather than talking to Kafka directly.

-Todd



On Tue, Nov 3, 2015 at 10:15 AM, Cassa L  wrote:

> Hi,
>  Has anyone used load balancers between publishers and  Kafka brokers? I
> want to do active-passive setup of Kafka in two datacenters.  My question
> is can I add GSLB layer between these two Kafka clusters to configure
> automatic fail over while publishing data?
>
> Thanks,
> LCassa
>

Re: Where is replication factor stored?

[Todd Palino]

Actually, be very careful with this. There are two different things stored
in Zookeeper, and depending on what you're interested in you want to make
sure you're looking at the right one.

If you want to know the replica assignment - that is, what brokers a given
partition is assigned to - you need to look at the following path:
/brokers/topics/(topic)

The data of that znode is JSON formatted, and the 'partitions' key is a
dictionary where they key is a string representation of the partition
number (i.e. it's not 0, it's "0") and the value is a list of the replicas
that the partition is assigned to. It's worth noting that this replica list
is also the order in which the preferred leader is selected (the first
replica in the list that is in sync will be selected as the leader).

If you want to know what the current in sync replicas are - that is, out of
the assigned replica list, which ones are currently considered to be in
sync, you need to look at the following path:
/brokers/topics/(topic)/partitions/(partition number)/state

The data of that znode is also JSON formatted, and the 'isr' key is a list
of the replicas that are currently considered to be in sync. The important
distinction here is that this list can be shorter than the actual assigned
replica list (from the znode above) if not all of the replicas are in sync.
The state znode also has a 'leader' key which holds the broker ID of the
replica that is currently the leader for that partition.

-Todd


On Fri, Oct 16, 2015 at 5:25 PM, Edward Ribeiro 
wrote:

> Hey, Guozhang,
>
> On Fri, Oct 16, 2015 at 6:20 PM, Guozhang Wang  wrote:
>
> > The replica list can be from at /brokers/topics//
> > partitions//state
> >
>
> Nice, good to know. Thanks! :)
>
> Regards,
> Edward​
>

Re: Where is replication factor stored?

[Todd Palino]

Sorry, I forgot the tl;dr on that :)

If you want to know the replication factor for a given partition, you want
to check the length of the replica list in the /brokers/topic/(topic) data
for that partition. Note that all the partitions for a topic do not have to
have the same replication factor (you can use partition reassignment to
change it). But if they are not all the same, some of the tooling will
break (such as altering the partition count for the topic).

-Todd


On Fri, Oct 16, 2015 at 5:39 PM, Todd Palino <tpal...@gmail.com> wrote:

> Actually, be very careful with this. There are two different things stored
> in Zookeeper, and depending on what you're interested in you want to make
> sure you're looking at the right one.
>
> If you want to know the replica assignment - that is, what brokers a given
> partition is assigned to - you need to look at the following path:
> /brokers/topics/(topic)
>
> The data of that znode is JSON formatted, and the 'partitions' key is a
> dictionary where they key is a string representation of the partition
> number (i.e. it's not 0, it's "0") and the value is a list of the replicas
> that the partition is assigned to. It's worth noting that this replica list
> is also the order in which the preferred leader is selected (the first
> replica in the list that is in sync will be selected as the leader).
>
> If you want to know what the current in sync replicas are - that is, out
> of the assigned replica list, which ones are currently considered to be in
> sync, you need to look at the following path:
> /brokers/topics/(topic)/partitions/(partition number)/state
>
> The data of that znode is also JSON formatted, and the 'isr' key is a list
> of the replicas that are currently considered to be in sync. The important
> distinction here is that this list can be shorter than the actual assigned
> replica list (from the znode above) if not all of the replicas are in sync.
> The state znode also has a 'leader' key which holds the broker ID of the
> replica that is currently the leader for that partition.
>
> -Todd
>
>
> On Fri, Oct 16, 2015 at 5:25 PM, Edward Ribeiro <edward.ribe...@gmail.com>
> wrote:
>
>> Hey, Guozhang,
>>
>> On Fri, Oct 16, 2015 at 6:20 PM, Guozhang Wang <wangg...@gmail.com>
>> wrote:
>>
>> > The replica list can be from at /brokers/topics//
>> > partitions//state
>> >
>>
>> Nice, good to know. Thanks! :)
>>
>> Regards,
>> Edward​
>>
>
>

Re: G1 tuning

[Todd Palino]

We've had no problems with G1 in all of our clusters with varying load
levels. I think we've seen an occasional long GC here and there, but
nothing recurring at this point.

What's the full command line that you're using with all the options?

-Todd


On Wed, Oct 14, 2015 at 2:18 PM, Scott Clasen  wrote:

> You can also use -Xmn with that gc to size the new gen such that those
> buffers don't get tenured
>
> I don't think that's an option with G1
>
> On Wednesday, October 14, 2015, Cory Kolbeck  wrote:
>
> > I'm not sure that will help here, you'll likely have the same
> > medium-lifetime buffers getting into the tenured generation and forcing
> > large collections.
> >
> > On Wed, Oct 14, 2015 at 10:00 AM, Gerrit Jansen van Vuuren <
> > gerrit...@gmail.com > wrote:
> >
> > > Hi,
> > >
> > > I've seen pauses using G1 in other applications and have found that
> > > -XX:+UseParallelGC
> > > -XX:+UseParallelOldGC  works best if you're having GC issues in general
> > on
> > > the JVM.
> > >
> > >
> > > Regards,
> > >  Gerrit
> > >
> > > On Wed, Oct 14, 2015 at 4:28 PM, Cory Kolbeck  > > wrote:
> > >
> > > > Hi folks,
> > > >
> > > > I'm a bit new to the operational side of G1, but pretty familiar with
> > its
> > > > basic concept. We recently set up a Kafka cluster to support a new
> > > product,
> > > > and are seeing some suboptimal GC performance. We're using the
> > parameters
> > > > suggested in the docs, except for having switched to java 1.8_40 in
> > order
> > > > to get better memory debugging. Even though the cluster is handling
> > only
> > > > 2-3k messages per second per node, we see periodic 11-18 second
> > > > stop-the-world pauses on a roughly hourly cadence. I've turned on
> > > > additional GC logging, and see no humongous allocations, it all seems
> > to
> > > be
> > > > buffers making it into the tenured gen. They appear to be
> collectable,
> > as
> > > > the collection triggered by dumping the heap collects them all. Ideas
> > for
> > > > additional diagnosis or tuning very welcome.
> > > >
> > > > --Cory
> > > >
> > >
> >
>

Re: Does Kafka recover all data if node is reimaged

[Todd Palino]

Yes. As long as you have not reassigned the partitions to other brokers,
the wiped broker will rebuild from replication. Keep in mind, however, that
if you are using the retention by time configuration, you will have 2x
retention on that broker for the length of retention. This means that if
your retention is 1 week, replication will copy over the last week's worth
of data. That data will not be expired for 1 week, as the expiration is
based on the file modification time.

There is work ongoing that will resolve this extra retention problem.

-Todd

On Monday, October 12, 2015, Rajasekar Elango 
wrote:

> I was wondering if a kafka broker node get reimaged and all data is wiped
> off, Will kafka recover all data on node from replication?
>
> --
> Thanks,
> Raja.
>

Re: Kafka availability guarantee

[Todd Palino]

To answer the question, yes, it is incorrect. There are a few things you
can do to minimize problems. One is to disable unclean leader election, use
acks=-1 on the producers, have an RF of 3 or greater, and set the min ISR
to 2. This means that the topic will only be available if there are at
least 2 replicas in sync, your producers will all wait for acknowledgements
from all in sync replicas (therefore, at least 2) before considering
produce requests to be complete, and if you get in a situation where all
three replicas go down, the cluster will not perform an unclean leader
election (which can lose messages).

Basically, you have to trade availability for correctness here. You get to
pick one.

-Todd


On Sun, Oct 11, 2015 at 5:10 PM,  wrote:

> You can enable unclean leader election, which would allow the lagging
> partition to still become leader. There would be some data loss (offsets
> between the leggy partition and the old leader) but the partition would
> stay online and available.
>
>
>
> Sent from my BlackBerry 10 smartphone on the TELUS network.
>   Original Message
> From: Elias Levy
> Sent: Sunday, October 11, 2015 5:00 PM
> To: users@kafka.apache.org
> Reply To: users@kafka.apache.org
> Subject: Kafka availability guarantee
>
> Reading through the Kafka documentation for statements regarding Kafka's
> availability guarantees one comes across this statement:
>
> *With this ISR model and f+1 replicas, a Kafka topic can tolerate f
> failures without losing committed messages.*
>
> In my opinion, this appears incorrect or at best misleading. Consider a
> partition with a replication factor of 3. If one of the replicas lags, but
> does not fail, the ISR will be shrank to a set of 2 replicas, the leader
> and and one follower. The leader will consider the message committed when
> itself and the in sync follower write the message to their respective
> logs. Where a concurrent failure of 2 nodes occur, specifically the
> failure of the leader and the in sync follower, there won't be any
> remaining in sync replicas to take over as leader without potential message
> loss. Therefore Kafka cannot tolerate any failure of *f* nodes, where *f*
> is N - 1 and N is the replication factor. Kafka can only tolerate a failure
> of *f* if we take N to be the ISR set size, which is a dynamic value and
> not a topic configuration parameter that can me set a priori. Kafka can
> tolerate some failures of *f* replicas when N is the replication factor, so
> long as at least one in sync replica survives, but it can't tolerate all
> such failures.
>
> Am I wrong?
>

Re: mapping events to topics

[Todd Palino]

Multiple topics is the model I would recommend for what you have described.
LinkedIn has an environment where we have a wide mix, in a lot of different
clusters. We have some topics that have one producer and one consumer
(queuing). We have some topics that are multi-producer (tracking and
metrics, mostly). Some of those are multi-consumer (tracking), and some are
mostly single consumer (metrics). Besides all of this, we have a couple
wildcard consumers that read everything (our audit system, and mirror
makers).

In your case, the rules engine sounds like a similar consumer case as our
audit consumer. I would not make the determination as to how many topics
you need based on that consumer because of that. Since the majority of what
you're describing is consumers who are interested in discrete data sets, go
with breaking out the topics based on that (all other things being equal).
While Gwen is absolutely right about her guidelines, consuming and throwing
away most of the data is a cardinal sin and should be avoided. Multi-topic
consumers are much less of a problem to deal with. Personally, I wouldn't
bother combining the messages into a separate topic for the rules engine -
I would just consume all the topics.

You mentioned message ordering, and that can present an issue. Now, you'd
likely have this problem regardless of how many topics you use, as ordering
is only guaranteed in a single partition. So you'd either have to have one
partition, or you would have to use some sort of partitioning scheme on the
messages that means hard ordering of all the messages matters less.
Obviously, when you have multiple topics it's the same as having multiple
partitions. You need to decide how important ordering within Kafka is to
your application, and if it can be handled separately inside of the
application.

-Todd



On Thu, Oct 8, 2015 at 8:50 AM, Mark Drago  wrote:

> Gwen,
>
> Thanks for your reply.  I understand all of the points you've made.  I
> think the challenge for us is that we have some consumers that are
> interested in messages of one type, but we also have a rules engine that is
> checking for events of many types and acting on them.
>
> If we put discrete event types on individual topics:
>   * Our rules engine would have to monitor many of these topics (10-20)
>   * Other consumers would only see messages they care about
>
> If we put all event types on one topic:
>   * Our rules engine only has to monitor one topic
>   * Other consumers would parse and then discard the majority of the
> messages that they see
>
> Perhaps a better approach would be to have different topics for the
> different use cases?  This would be similar to an approach that merges
> smaller topics together as needed.  So, each event type would be on its own
> topic but then we would put a subset of those messages on another topic
> that is destined for the rules engine.  The consumers that only care about
> one message type would listen on dedicated topics and the rules engine
> would just monitor one topic for all of the events that it cares about.  We
> would need to have something moving/creating messages on the rules engine
> topic.  We may also run in to another set of problems as the ordering of
> messages of different types no longer exists as they're coming from
> separate topics.
>
> I'm curious to hear if anyone else has been in a similar situation and had
> to make a judgement call about the best approach to take.
>
> Thanks,
> Mark.
>
> I usually approach this questions by looking at possible consumers. You
> > usually want each consumer to read from relatively few topics, use most
> > of the messages it receives and have fairly cohesive logic for using
> these
> > messages. Signs that things went wrong with too few topics:
> > * Consumers that throw away 90% of the messages on topics they read
> > * Consumers with gigantic switch statements for handling all the
> different
> > message types they get Signs that you have too many topics:
> > * Every consumer needs to read messages from 20 different topics in order
> > to construct the objects it actually uses If you ever did data modeling
> > for a datawarehouse, this will look a bit
> > familiar :) Gwen
> > On Tue, Oct 6, 2015 at 4:46 PM, Mark Drago  wrote:
> >
> > Hello,
> > >
> > > At my organization we are already using kafka in a few areas, but we're
> > > looking to expand our use and we're struggling with how best to
> > distribute
> > > our events on to topics.
> > >
> > > We have on the order of 30 different kinds of events that we'd like to
> > > distribute via kafka. We have one or two consumers that have a need to
> > > consume many of these types of events (~20 out of the 30) and we have
> > other
> > > consumers that are only interested in one type of event.
> > >
> > > We're trying to decide between a model where we have one topic
> containing
> > > many kinds of events or a model where we have many topics each

Re: Log Cleaner Thread Stops

[Todd Palino]

This is correct, compression isn't used for the offsets at all. If, for
some reason, you do have either a compressed or a corrupt message somewhere
in the topic, the method I mentioned previously will flush it out. We
haven't seen that as a recurring problem, so fixing it once is sufficient.

-Todd


On Mon, Sep 28, 2015 at 9:53 AM, Jason Rosenberg <j...@squareup.com> wrote:

> Just to clarify too, if the only use case for log-compaction we use is for
> the __consumer_offsets, we should be ok, correct?  I assume compression is
> not used by default for consumer offsets?
>
> Jason
>
> On Fri, Sep 25, 2015 at 12:15 AM, Todd Palino <tpal...@gmail.com> wrote:
>
> > For now, that's the way it is. Historically, we've only monitored the lag
> > for our infrastructure applications. Other users are responsible for
> their
> > own checking, typically using the maxlag mbean or some application
> specific
> > metric. Besides the core, we've probably got a dozen or so consumers
> moved
> > over to Kafka committed offsets at this point.
> >
> > Of course, just those apps do cover well over a hundred consumer groups
> :)
> >
> > -Todd
> >
> > On Thursday, September 24, 2015, James Cheng <jch...@tivo.com> wrote:
> >
> > >
> > > > On Sep 24, 2015, at 8:11 PM, Todd Palino <tpal...@gmail.com
> > > <javascript:;>> wrote:
> > > >
> > > > Well, in general you can't currently use compressed messages in any
> > topic
> > > > that has compaction turned on regardless of whether or not you are
> > using
> > > > Kafka-committed offsets. The log compaction thread will die either
> way.
> > > > There's only one compression thread for the broker that runs on all
> > > topics
> > > > that use compaction.
> > > >
> > > > Jason, to address your question, it's probably wise to wait for now.
> > > > Zookeeper offsets work, so unless it's broke, don't fix it for now.
> > We're
> > > > using Kafka-committed offsets at LinkedIn for our mirror makers and
> our
> > > > auditor application (both of which are considered infrastructure
> > > > applications for Kafka), but we're not encouraging other internal
> users
> > > to
> > > > switch over just yet.
> > > >
> > >
> > > Burrow depends on kafka-commited offsets, doesn’t it? I guess that
> means
> > > Burrow is only being used to monitor your mirror makers and auditor
> > > application, then?
> > >
> > > -James
> > >
> > > > -Todd
> > > >
> > > >
> > > > On Wed, Sep 23, 2015 at 3:21 PM, James Cheng <jch...@tivo.com
> > > <javascript:;>> wrote:
> > > >
> > > >>
> > > >> On Sep 18, 2015, at 10:25 AM, Todd Palino <tpal...@gmail.com
> > > <javascript:;>> wrote:
> > > >>
> > > >>> I think the last major issue with log compaction (that it couldn't
> > > handle
> > > >>> compressed messages) was committed as part of
> > > >>> https://issues.apache.org/jira/browse/KAFKA-1374 in August, but
> I'm
> > > not
> > > >>> certain what version this will end up in. It may be part of
> 0.8.2.2.
> > > >>>
> > > >>> Regardless, you'll probably be OK now. We've found that once we
> clean
> > > >> this
> > > >>> issue up once it doesn't appear to recur. As long as you're not
> > writing
> > > >> in
> > > >>> compressed messages to a log compacted topic (and that won't happen
> > > with
> > > >>> __consumer_offsets, as it's managed by the brokers themselves - it
> > > would
> > > >>> only be if you were using other log compacted topics), you're
> likely
> > in
> > > >> the
> > > >>> clear now.
> > > >>>
> > > >>
> > > >> Todd,
> > > >>
> > > >> If I understand your description of the problem, you are saying that
> > > >> enabling log compaction on a topic with compressed messages can
> > (will?)
> > > >> cause the log cleaner to crash when it encounters those compressed
> > > >> messages. And the death of the cleaner thread will prevent log
> > > compaction
> > > >> from running on other topics, even ones that don't have compressed
> > > messages.
> > > >>
> >

Re: Frequent Consumer and Producer Disconnects

[Todd Palino]

Topic creation should only cause a rebalance for wildcard consumers (and I
believe that is regardless of whether or not the wildcard covers the topic
- once the ZK watch fires a rebalance is going to happen).

Back to the original concern, it would be helpful to see more of that log,
in that case. When a rebalance is triggered, there will be a log message
that will indicate why. This is going to be caused by a change in the group
membership (which has a number of causes, but at least it narrows it down)
or a topic change. Figuring out why the consumers are rebalancing is the
first step to trying to reduce it.

-Todd


On Saturday, September 26, 2015, noah <iamn...@gmail.com> wrote:

> Thanks, that gives us some more to look at.
>
> That is unfortunately a small section of the log file. When we hit this
> problem (which is not every time,) it will continue like that for hours.
>
> We also still have developers creating topics semi-regularly, which it
> seems like can cause the high level consumer to disconnect?
>
>
> On Fri, Sep 25, 2015 at 6:16 PM Todd Palino <tpal...@gmail.com
> <javascript:_e(%7B%7D,'cvml','tpal...@gmail.com');>> wrote:
>
>> That rebalance cycle doesn't look endless. I see that you started 23
>> consumers, and I see 23 rebalances finishing successfully, which is
>> correct. You will see rebalance messages from all of the consumers you
>> started. It all happens within about 2 seconds, which is fine. I agree that
>> there is a lot of log messages, but I'm not seeing anything that is
>> particularly a problem here. After the segment of pot you provided, your
>> consumers will be running properly. Now, given you have a topic with 16
>> partitions, and you're running 23 consumers, 7 of those consumer threads
>> are going to be idle because they do not own partitions.
>>
>> -Todd
>>
>>
>> On Fri, Sep 25, 2015 at 3:27 PM, noah <iamn...@gmail.com
>> <javascript:_e(%7B%7D,'cvml','iamn...@gmail.com');>> wrote:
>>
>>> We're seeing this the most on developer machines that are starting up
>>> multiple high level consumers on the same topic+group as part of service
>>> startup. The consumers do not seem to get a chance to consume anything
>>> before they disconnect.
>>>
>>> These are developer topics, so it is possible/likely that there isn't
>>> anything for them to consume in the topic, but the same service will start
>>> producing, so I would expect them to not be idle for long.
>>>
>>> Could it be the way we are bring up multiple consumers at the same time
>>> is hitting some sort of endless rebalance cycle? And/or the resulting
>>> thrashing is causing them to time out, rebalance, etc.?
>>>
>>> I've tried attaching the logs again. Thanks!
>>>
>>> On Fri, Sep 25, 2015 at 3:33 PM Todd Palino <tpal...@gmail.com
>>> <javascript:_e(%7B%7D,'cvml','tpal...@gmail.com');>> wrote:
>>>
>>>> I don't see the logs attached, but what does the GC look like in your
>>>> applications? A lot of times this is caused (at least on the consumer
>>>> side)
>>>> by the Zookeeper session expiring due to excessive GC activity, which
>>>> causes the consumers to go into a rebalance and change up their
>>>> connections.
>>>>
>>>> -Todd
>>>>
>>>>
>>>> On Fri, Sep 25, 2015 at 1:25 PM, Gwen Shapira <g...@confluent.io
>>>> <javascript:_e(%7B%7D,'cvml','g...@confluent.io');>> wrote:
>>>>
>>>> > How busy are the clients?
>>>> >
>>>> > The brokers occasionally close idle connections, this is normal and
>>>> > typically not something to worry about.
>>>> > However, this shouldn't happen to consumers that are actively reading
>>>> data.
>>>> >
>>>> > I'm wondering if the "consumers not making any progress" could be due
>>>> to a
>>>> > different issue, and because they are idle, the connection closes (vs
>>>> the
>>>> > other way around).
>>>> >
>>>> > On Thu, Sep 24, 2015 at 2:32 PM, noah <iamn...@gmail.com
>>>> <javascript:_e(%7B%7D,'cvml','iamn...@gmail.com');>> wrote:
>>>> >
>>>> > > We are having issues with producers and consumers frequently fully
>>>> > > disconnecting (from both the brokers and ZK) and reconnecting
>>>> without any
>>>> > > apparent cause. On our production systems it can happen anywhere
>>>> from
>>>> > every
>>>> > > 10-15 seconds to 15-20 minutes. On our less beefy test systems and
>>>> > > developer laptops, it can happen almost constantly.
>>>> > >
>>>> > > We see no errors in the logs (sample attached), just a message for
>>>> each
>>>> > of
>>>> > > our our consumers and producers disconnecting, then reconnecting.
>>>> On the
>>>> > > systems where it happens constantly, the consumers are not making
>>>> any
>>>> > > progress.
>>>> > >
>>>> > > The logs on the brokers are equally unhelpful, they show only
>>>> frequent
>>>> > > connects and reconnects, without any apparent cause.
>>>> > >
>>>> > > What could be causing this behavior?
>>>> > >
>>>> > >
>>>> >
>>>>
>>>
>>

Re: Frequent Consumer and Producer Disconnects

[Todd Palino]

I don't see the logs attached, but what does the GC look like in your
applications? A lot of times this is caused (at least on the consumer side)
by the Zookeeper session expiring due to excessive GC activity, which
causes the consumers to go into a rebalance and change up their connections.

-Todd


On Fri, Sep 25, 2015 at 1:25 PM, Gwen Shapira  wrote:

> How busy are the clients?
>
> The brokers occasionally close idle connections, this is normal and
> typically not something to worry about.
> However, this shouldn't happen to consumers that are actively reading data.
>
> I'm wondering if the "consumers not making any progress" could be due to a
> different issue, and because they are idle, the connection closes (vs the
> other way around).
>
> On Thu, Sep 24, 2015 at 2:32 PM, noah  wrote:
>
> > We are having issues with producers and consumers frequently fully
> > disconnecting (from both the brokers and ZK) and reconnecting without any
> > apparent cause. On our production systems it can happen anywhere from
> every
> > 10-15 seconds to 15-20 minutes. On our less beefy test systems and
> > developer laptops, it can happen almost constantly.
> >
> > We see no errors in the logs (sample attached), just a message for each
> of
> > our our consumers and producers disconnecting, then reconnecting. On the
> > systems where it happens constantly, the consumers are not making any
> > progress.
> >
> > The logs on the brokers are equally unhelpful, they show only frequent
> > connects and reconnects, without any apparent cause.
> >
> > What could be causing this behavior?
> >
> >
>

Re: Frequent Consumer and Producer Disconnects

[Todd Palino]

That rebalance cycle doesn't look endless. I see that you started 23
consumers, and I see 23 rebalances finishing successfully, which is
correct. You will see rebalance messages from all of the consumers you
started. It all happens within about 2 seconds, which is fine. I agree that
there is a lot of log messages, but I'm not seeing anything that is
particularly a problem here. After the segment of pot you provided, your
consumers will be running properly. Now, given you have a topic with 16
partitions, and you're running 23 consumers, 7 of those consumer threads
are going to be idle because they do not own partitions.

-Todd


On Fri, Sep 25, 2015 at 3:27 PM, noah <iamn...@gmail.com> wrote:

> We're seeing this the most on developer machines that are starting up
> multiple high level consumers on the same topic+group as part of service
> startup. The consumers do not seem to get a chance to consume anything
> before they disconnect.
>
> These are developer topics, so it is possible/likely that there isn't
> anything for them to consume in the topic, but the same service will start
> producing, so I would expect them to not be idle for long.
>
> Could it be the way we are bring up multiple consumers at the same time is
> hitting some sort of endless rebalance cycle? And/or the resulting
> thrashing is causing them to time out, rebalance, etc.?
>
> I've tried attaching the logs again. Thanks!
>
> On Fri, Sep 25, 2015 at 3:33 PM Todd Palino <tpal...@gmail.com> wrote:
>
>> I don't see the logs attached, but what does the GC look like in your
>> applications? A lot of times this is caused (at least on the consumer
>> side)
>> by the Zookeeper session expiring due to excessive GC activity, which
>> causes the consumers to go into a rebalance and change up their
>> connections.
>>
>> -Todd
>>
>>
>> On Fri, Sep 25, 2015 at 1:25 PM, Gwen Shapira <g...@confluent.io> wrote:
>>
>> > How busy are the clients?
>> >
>> > The brokers occasionally close idle connections, this is normal and
>> > typically not something to worry about.
>> > However, this shouldn't happen to consumers that are actively reading
>> data.
>> >
>> > I'm wondering if the "consumers not making any progress" could be due
>> to a
>> > different issue, and because they are idle, the connection closes (vs
>> the
>> > other way around).
>> >
>> > On Thu, Sep 24, 2015 at 2:32 PM, noah <iamn...@gmail.com> wrote:
>> >
>> > > We are having issues with producers and consumers frequently fully
>> > > disconnecting (from both the brokers and ZK) and reconnecting without
>> any
>> > > apparent cause. On our production systems it can happen anywhere from
>> > every
>> > > 10-15 seconds to 15-20 minutes. On our less beefy test systems and
>> > > developer laptops, it can happen almost constantly.
>> > >
>> > > We see no errors in the logs (sample attached), just a message for
>> each
>> > of
>> > > our our consumers and producers disconnecting, then reconnecting. On
>> the
>> > > systems where it happens constantly, the consumers are not making any
>> > > progress.
>> > >
>> > > The logs on the brokers are equally unhelpful, they show only frequent
>> > > connects and reconnects, without any apparent cause.
>> > >
>> > > What could be causing this behavior?
>> > >
>> > >
>> >
>>
>

Re: Log Cleaner Thread Stops

[Todd Palino]

Well, in general you can't currently use compressed messages in any topic
that has compaction turned on regardless of whether or not you are using
Kafka-committed offsets. The log compaction thread will die either way.
There's only one compression thread for the broker that runs on all topics
that use compaction.

Jason, to address your question, it's probably wise to wait for now.
Zookeeper offsets work, so unless it's broke, don't fix it for now. We're
using Kafka-committed offsets at LinkedIn for our mirror makers and our
auditor application (both of which are considered infrastructure
applications for Kafka), but we're not encouraging other internal users to
switch over just yet.

-Todd


On Wed, Sep 23, 2015 at 3:21 PM, James Cheng <jch...@tivo.com> wrote:

>
> On Sep 18, 2015, at 10:25 AM, Todd Palino <tpal...@gmail.com> wrote:
>
> > I think the last major issue with log compaction (that it couldn't handle
> > compressed messages) was committed as part of
> > https://issues.apache.org/jira/browse/KAFKA-1374 in August, but I'm not
> > certain what version this will end up in. It may be part of 0.8.2.2.
> >
> > Regardless, you'll probably be OK now. We've found that once we clean
> this
> > issue up once it doesn't appear to recur. As long as you're not writing
> in
> > compressed messages to a log compacted topic (and that won't happen with
> > __consumer_offsets, as it's managed by the brokers themselves - it would
> > only be if you were using other log compacted topics), you're likely in
> the
> > clear now.
> >
>
> Todd,
>
> If I understand your description of the problem, you are saying that
> enabling log compaction on a topic with compressed messages can (will?)
> cause the log cleaner to crash when it encounters those compressed
> messages. And the death of the cleaner thread will prevent log compaction
> from running on other topics, even ones that don't have compressed messages.
>
> That means if we have a cluster where we want to use log compaction on
> *any* topic, we need to either:
> 1) apply https://issues.apache.org/jira/browse/KAFKA-1374 (or upgrade to
> some version it is applied)
> OR
> 2) make sure that we don't use compressed messages in *any* topic that has
> log compaction turned on.
>
> And, more specifically, if we want to make use of __consumer_offsets, then
> we cannot use compressed messages in any topic that has compaction turned
> on.
>
> Is that right?
> -James
>
> > -Todd
> >
> >
> > On Fri, Sep 18, 2015 at 9:54 AM, John Holland <
> > john.holl...@objectpartners.com> wrote:
> >
> >> Thanks!
> >>
> >> I did what you suggested and it worked except it was necessary for me to
> >> remove the cleaner-offset-checkpoint file from the data directory and
> >> restart the servers.  The log indicates all is well.
> >>
> >> Do you know what version the fix to this will be in? I'm not looking
> >> forward to dealing with this on a reoccurring basis.
> >>
> >> -John
> >>
> >> On Fri, Sep 18, 2015 at 8:48 AM Todd Palino <tpal...@gmail.com> wrote:
> >>
> >>> Yes, this is a known concern, and it should be fixed with recent
> commits.
> >>> In the meantime, you'll have to do a little manual cleanup.
> >>>
> >>> The problem you're running into is a corrupt message in the offsets
> >> topic.
> >>> We've seen this a lot. What you need to do is set the topic
> configuration
> >>> to remove the cleanup.policy config, and set retention.ms and
> segment.ms
> >>> to
> >>> something reasonably low. I suggest using a value of 3 or 4 times your
> >>> commit interval for consumers. Then wait until the log segments are
> >> reaped
> >>> (wait twice as long as the retention.ms you chose, to be safe). Once
> >> this
> >>> is done, you can set the topic configuration back the way it was
> (remove
> >>> segment.ms and retention.ms configs, and set cleanup.policy=compact).
> >>> Lastly, you'll need to do a rolling bounce of the cluster to restart
> the
> >>> brokers (which restarts the log cleaner threads). Technically, you only
> >>> need to restart brokers where the threads have died, but it's easier to
> >>> just restart all of them.
> >>>
> >>> Keep in mind that when you do this, you are deleting old offsets. If
> your
> >>> consumers are all live and healthy, this shouldn't be a problem because
> >>> they will just continue to commit their offsets properly. But if you
> have
> >

Re: Log Cleaner Thread Stops

[Todd Palino]

For now, that's the way it is. Historically, we've only monitored the lag
for our infrastructure applications. Other users are responsible for their
own checking, typically using the maxlag mbean or some application specific
metric. Besides the core, we've probably got a dozen or so consumers moved
over to Kafka committed offsets at this point.

Of course, just those apps do cover well over a hundred consumer groups :)

-Todd

On Thursday, September 24, 2015, James Cheng <jch...@tivo.com> wrote:

>
> > On Sep 24, 2015, at 8:11 PM, Todd Palino <tpal...@gmail.com
> <javascript:;>> wrote:
> >
> > Well, in general you can't currently use compressed messages in any topic
> > that has compaction turned on regardless of whether or not you are using
> > Kafka-committed offsets. The log compaction thread will die either way.
> > There's only one compression thread for the broker that runs on all
> topics
> > that use compaction.
> >
> > Jason, to address your question, it's probably wise to wait for now.
> > Zookeeper offsets work, so unless it's broke, don't fix it for now. We're
> > using Kafka-committed offsets at LinkedIn for our mirror makers and our
> > auditor application (both of which are considered infrastructure
> > applications for Kafka), but we're not encouraging other internal users
> to
> > switch over just yet.
> >
>
> Burrow depends on kafka-commited offsets, doesn’t it? I guess that means
> Burrow is only being used to monitor your mirror makers and auditor
> application, then?
>
> -James
>
> > -Todd
> >
> >
> > On Wed, Sep 23, 2015 at 3:21 PM, James Cheng <jch...@tivo.com
> <javascript:;>> wrote:
> >
> >>
> >> On Sep 18, 2015, at 10:25 AM, Todd Palino <tpal...@gmail.com
> <javascript:;>> wrote:
> >>
> >>> I think the last major issue with log compaction (that it couldn't
> handle
> >>> compressed messages) was committed as part of
> >>> https://issues.apache.org/jira/browse/KAFKA-1374 in August, but I'm
> not
> >>> certain what version this will end up in. It may be part of 0.8.2.2.
> >>>
> >>> Regardless, you'll probably be OK now. We've found that once we clean
> >> this
> >>> issue up once it doesn't appear to recur. As long as you're not writing
> >> in
> >>> compressed messages to a log compacted topic (and that won't happen
> with
> >>> __consumer_offsets, as it's managed by the brokers themselves - it
> would
> >>> only be if you were using other log compacted topics), you're likely in
> >> the
> >>> clear now.
> >>>
> >>
> >> Todd,
> >>
> >> If I understand your description of the problem, you are saying that
> >> enabling log compaction on a topic with compressed messages can (will?)
> >> cause the log cleaner to crash when it encounters those compressed
> >> messages. And the death of the cleaner thread will prevent log
> compaction
> >> from running on other topics, even ones that don't have compressed
> messages.
> >>
> >> That means if we have a cluster where we want to use log compaction on
> >> *any* topic, we need to either:
> >> 1) apply https://issues.apache.org/jira/browse/KAFKA-1374 (or upgrade
> to
> >> some version it is applied)
> >> OR
> >> 2) make sure that we don't use compressed messages in *any* topic that
> has
> >> log compaction turned on.
> >>
> >> And, more specifically, if we want to make use of __consumer_offsets,
> then
> >> we cannot use compressed messages in any topic that has compaction
> turned
> >> on.
> >>
> >> Is that right?
> >> -James
> >>
> >>> -Todd
> >>>
> >>>
> >>> On Fri, Sep 18, 2015 at 9:54 AM, John Holland <
> >>> john.holl...@objectpartners.com <javascript:;>> wrote:
> >>>
> >>>> Thanks!
> >>>>
> >>>> I did what you suggested and it worked except it was necessary for me
> to
> >>>> remove the cleaner-offset-checkpoint file from the data directory and
> >>>> restart the servers.  The log indicates all is well.
> >>>>
> >>>> Do you know what version the fix to this will be in? I'm not looking
> >>>> forward to dealing with this on a reoccurring basis.
> >>>>
> >>>> -John
> >>>>
> >>>> On Fri, Sep 18, 2015 at 8:48 AM Todd Palino <tpal...@gmai

Re: log.retention.hours not working?

[Todd Palino]

Retention is going to be based on a combination of both the retention and
segment size settings (as a side note, it's recommended to use
log.retention.ms and log.segment.ms, not the hours config. That's there for
legacy reasons, but the ms configs are more consistent). As messages are
received by Kafka, they are written to the current open log segment for
each partition. That segment is rotated when either the log.segment.bytes
or the log.segment.ms limit is reached. Once that happens, the log segment
is closed and a new one is opened. Only after a log segment is closed can
it be deleted via the retention settings. Once the log segment is closed
AND either all the messages in the segment are older than log.retention.ms
OR the total partition size is greater than log.retention.bytes, then the
log segment is purged.

As a note, the default segment limit is 1 gibibyte. So if you've only
written in 1k of messages, you have a long way to go before that segment
gets rotated. This is why the retention is referred to as a minimum time.
You can easily retain much more than you're expecting for slow topics.

-Todd


On Mon, Sep 21, 2015 at 7:28 PM, allen chan 
wrote:

> I guess that kind of makes sense.
> The following section in the config is what confused me:
> *"# The following configurations control the disposal of log segments. The
> policy can*
> *# be set to delete segments after a period of time, or after a given size
> has accumulated.*
> *# A segment will be deleted whenever *either* of these criteria are met.
> Deletion always happens*
> *# from the end of the log."*
>
> That makes it sound like deletion will happen if either of the criteria is
> met.
> I thought the whole idea of those two settings (time and bytes) is telling
> the application when it will need to delete.
>
>
>
> On Mon, Sep 21, 2015 at 7:10 PM, noah  wrote:
>
> > "minimum age of a log file to be eligible for deletion" Key word is
> > minimum. If you only have 1k logs, Kafka doesn't need to delete anything.
> > Try to push more data through and when it needs to, it will start
> deleting
> > old logs.
> >
> > On Mon, Sep 21, 2015 at 8:58 PM allen chan  >
> > wrote:
> >
> > > Hi,
> > >
> > > Just brought up new kafka cluster for testing.
> > > Was able to use the console producers to send 1k of logs and received
> it
> > on
> > > the console consumer side.
> > >
> > > The one issue that i have right now is that the retention period does
> not
> > > seem to be working.
> > >
> > > *# The minimum age of a log file to be eligible for deletion*
> > > *log.retention.hours=1*
> > >
> > > I have waited for almost 2 hours and the 1k of logs are still in kafka.
> > >
> > > I did see these messages pop up on the console
> > > *[2015-09-21 17:12:01,236] INFO Scheduling log segment 0 for log test-1
> > for
> > > deletion. (kafka.log.Log)*
> > > *[2015-09-21 17:13:01,238] INFO Deleting segment 0 from log test-1.
> > > (kafka.log.Log)*
> > > *[2015-09-21 17:13:01,239] INFO Deleting index
> > > /var/log/kafka/test-1/.index.deleted
> > > (kafka.log.OffsetIndex)*
> > >
> > > I know the logs are still in there because i am using
> > > the kafka-consumer-offset-checker.sh and it says how many messages the
> > > logSize is.
> > >
> > > What am i missing in my configuration?
> > >
> > >
> > >
> > > Thanks!
> > >
> > > --
> > > Allen Michael Chan
> > >
> >
>
>
>
> --
> Allen Michael Chan
>

Re: Log Cleaner Thread Stops

[Todd Palino]

Yes, this is a known concern, and it should be fixed with recent commits.
In the meantime, you'll have to do a little manual cleanup.

The problem you're running into is a corrupt message in the offsets topic.
We've seen this a lot. What you need to do is set the topic configuration
to remove the cleanup.policy config, and set retention.ms and segment.ms to
something reasonably low. I suggest using a value of 3 or 4 times your
commit interval for consumers. Then wait until the log segments are reaped
(wait twice as long as the retention.ms you chose, to be safe). Once this
is done, you can set the topic configuration back the way it was (remove
segment.ms and retention.ms configs, and set cleanup.policy=compact).
Lastly, you'll need to do a rolling bounce of the cluster to restart the
brokers (which restarts the log cleaner threads). Technically, you only
need to restart brokers where the threads have died, but it's easier to
just restart all of them.

Keep in mind that when you do this, you are deleting old offsets. If your
consumers are all live and healthy, this shouldn't be a problem because
they will just continue to commit their offsets properly. But if you have
an offline consumer, you'll lose the committed offsets by doing this.

-Todd


On Fri, Sep 18, 2015 at 5:31 AM, John Holland <
john.holl...@objectpartners.com> wrote:

> I've been experiencing this issue across several of our environments ever
> since we enabled the log cleaner for the __consumer_offsets topic.
>
> We are on version 0.8.2.1 of kafka, using the new producer.  All of our
> consumers are set to commit to kafka only.
>
> Below is the stack trace in the log I've encountered across several
> different clusters.  A simple restart of kafka will allow compaction to
> continue on all of the other partitions but the incorrect one will always
> fail.
>
> Here are the values for it from the kafka-topics --describe command:
>
> Topic:__consumer_offsets PartitionCount:50 ReplicationFactor:3
> Configs:segment.bytes=104857600,cleanup.policy=compact
>
> Are there any recommendations on how to prevent this and the best way to
> recover from this exception?  This is causing disk space to fill up quickly
> on the node.
>
> I did see an open issue that seems very similar to this
> https://issues.apache.org/jira/browse/KAFKA-1641 but this is the
> __consumer_offsets topic which I have not had any part in setting up nor
> producing to.
>
> [2015-09-18 02:57:25,520] INFO Cleaner 0: Beginning cleaning of log
> __consumer_offsets-17. (kafka.log.LogCleaner)
> [2015-09-18 02:57:25,520] INFO Cleaner 0: Building offset map for
> __consumer_offsets-17... (kafka.log.LogCleaner)
> [2015-09-18 02:57:25,609] INFO Cleaner 0: Building offset map for log
> __consumer_offsets-17 for 46 segments in offset range [468079184,
> 528707475). (kafka.log.LogCleaner)
> [2015-09-18 02:57:25,645] ERROR [kafka-log-cleaner-thread-0], Error due to
>  (kafka.log.LogCleaner)
> java.lang.IllegalArgumentException: requirement failed: Last clean offset
> is 468079184 but segment base offset is 0 for log __consumer_offsets-17.
> at scala.Predef$.require(Predef.scala:233)
> at kafka.log.Cleaner.buildOffsetMap(LogCleaner.scala:509)
> at kafka.log.Cleaner.clean(LogCleaner.scala:307)
> at
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:221)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:199)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
> [2015-09-18 02:57:25,654] INFO [kafka-log-cleaner-thread-0], Stopped
>  (kafka.log.LogCleaner)
>
> -John
>

Re: Log Cleaner Thread Stops

[Todd Palino]

I think the last major issue with log compaction (that it couldn't handle
compressed messages) was committed as part of
https://issues.apache.org/jira/browse/KAFKA-1374 in August, but I'm not
certain what version this will end up in. It may be part of 0.8.2.2.

Regardless, you'll probably be OK now. We've found that once we clean this
issue up once it doesn't appear to recur. As long as you're not writing in
compressed messages to a log compacted topic (and that won't happen with
__consumer_offsets, as it's managed by the brokers themselves - it would
only be if you were using other log compacted topics), you're likely in the
clear now.

-Todd


On Fri, Sep 18, 2015 at 9:54 AM, John Holland <
john.holl...@objectpartners.com> wrote:

> Thanks!
>
> I did what you suggested and it worked except it was necessary for me to
> remove the cleaner-offset-checkpoint file from the data directory and
> restart the servers.  The log indicates all is well.
>
> Do you know what version the fix to this will be in? I'm not looking
> forward to dealing with this on a reoccurring basis.
>
> -John
>
> On Fri, Sep 18, 2015 at 8:48 AM Todd Palino <tpal...@gmail.com> wrote:
>
> > Yes, this is a known concern, and it should be fixed with recent commits.
> > In the meantime, you'll have to do a little manual cleanup.
> >
> > The problem you're running into is a corrupt message in the offsets
> topic.
> > We've seen this a lot. What you need to do is set the topic configuration
> > to remove the cleanup.policy config, and set retention.ms and segment.ms
> > to
> > something reasonably low. I suggest using a value of 3 or 4 times your
> > commit interval for consumers. Then wait until the log segments are
> reaped
> > (wait twice as long as the retention.ms you chose, to be safe). Once
> this
> > is done, you can set the topic configuration back the way it was (remove
> > segment.ms and retention.ms configs, and set cleanup.policy=compact).
> > Lastly, you'll need to do a rolling bounce of the cluster to restart the
> > brokers (which restarts the log cleaner threads). Technically, you only
> > need to restart brokers where the threads have died, but it's easier to
> > just restart all of them.
> >
> > Keep in mind that when you do this, you are deleting old offsets. If your
> > consumers are all live and healthy, this shouldn't be a problem because
> > they will just continue to commit their offsets properly. But if you have
> > an offline consumer, you'll lose the committed offsets by doing this.
> >
> > -Todd
> >
> >
> > On Fri, Sep 18, 2015 at 5:31 AM, John Holland <
> > john.holl...@objectpartners.com> wrote:
> >
> > > I've been experiencing this issue across several of our environments
> ever
> > > since we enabled the log cleaner for the __consumer_offsets topic.
> > >
> > > We are on version 0.8.2.1 of kafka, using the new producer.  All of our
> > > consumers are set to commit to kafka only.
> > >
> > > Below is the stack trace in the log I've encountered across several
> > > different clusters.  A simple restart of kafka will allow compaction to
> > > continue on all of the other partitions but the incorrect one will
> always
> > > fail.
> > >
> > > Here are the values for it from the kafka-topics --describe command:
> > >
> > > Topic:__consumer_offsets PartitionCount:50 ReplicationFactor:3
> > > Configs:segment.bytes=104857600,cleanup.policy=compact
> > >
> > > Are there any recommendations on how to prevent this and the best way
> to
> > > recover from this exception?  This is causing disk space to fill up
> > quickly
> > > on the node.
> > >
> > > I did see an open issue that seems very similar to this
> > > https://issues.apache.org/jira/browse/KAFKA-1641 but this is the
> > > __consumer_offsets topic which I have not had any part in setting up
> nor
> > > producing to.
> > >
> > > [2015-09-18 02:57:25,520] INFO Cleaner 0: Beginning cleaning of log
> > > __consumer_offsets-17. (kafka.log.LogCleaner)
> > > [2015-09-18 02:57:25,520] INFO Cleaner 0: Building offset map for
> > > __consumer_offsets-17... (kafka.log.LogCleaner)
> > > [2015-09-18 02:57:25,609] INFO Cleaner 0: Building offset map for log
> > > __consumer_offsets-17 for 46 segments in offset range [468079184,
> > > 528707475). (kafka.log.LogCleaner)
> > > [2015-09-18 02:57:25,645] ERROR [kafka-log-cleaner-thread-0], Error due
> > to
> > >  (kafka.log.LogCleaner)
> > > java.lang

Re: Delay in Zookeeper Offset updates

[Todd Palino]

Consumer offsets in Zookeeper are not handled by the Kafka brokers at all -
the consumer writes those directly to Zookeeper. Most likely, what you are
seeing is the interval over which the consumer is committing offsets.
Assuming that you are using the auto.commit.enable setting (it defaults to
true, so unless you have changed that in the consumer configuration it will
apply here), you should check the setting of auto.commit.interval for your
consumer. I believe the default is still 60 seconds (or a config value of
6). If you're seeing the values change in ZK every 5-6 seconds, then it
sounds like it is set much lower.

The setting for the offset commit interval is entirely up to you. The lower
the value, the more often offsets are checkpointed to Zookeeper. However
that means that your consumer is writing to Zookeeper more often, and that
has performance impacts. In an environment like ours, where there is a
large number of consumers for a given cluster, we actually had to increase
the interval to 10 minutes for quite some time because the load on
Zookeeper was just too high. Once we moved the ZK transaction logs to SSDs,
we were able to drop that back to 60 seconds.

-Todd



On Thu, Sep 17, 2015 at 1:14 PM, nitin sharma 
wrote:

> anyone faced this issue?
>
> Regards,
> Nitin Kumar Sharma.
>
>
> On Wed, Sep 2, 2015 at 5:32 PM, nitin sharma 
> wrote:
>
> > Hi All,
> >
> > i have run into a weird issue with my Kafka setup.
> >
> > I see that it takes around 5-6 sec for Zookeeper to update the offset for
> > Kafka topics.
> >
> > I am running "ConsumerOffsetChecker" tool to see that lag and what i
> found
> > that even when my consumer is not up, it takes 5-6 sec for Zookeeper to
> > show the updated Offset .
> >
> > Is this behavior can be fixed? I have tried adding following parameters
> in
> > my Kafka server.property files but no luck.
> >
> > log.flush.interval.messages=1
> > log.default.flush.scheduler.interval.ms=500
> >
> >
> > ConsumerOffsetChecker:
> >  bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect
> > localhost:2181 --group perfConsumer
> >
> >
> > Regards,
> > Nitin Kumar Sharma.
> >
> >
>

Re: Closing connection messages

[Todd Palino]

This message is regarding a normal connection close. You see it in the logs
for any connection - consumers, producers, replica fetchers. It can be
particularly noisy because metadata requests often happen on their own
connection.

The log message has been moved to debug level in recent commits (it
actually was changed a while back but snuck back in with one of the larger
sets of changes). In general, I'd say you should just ignore it for now.

-Todd


On Wed, Sep 16, 2015 at 9:55 PM, Rajiv Kurian  wrote:

> My broker logs are full of messages of the following type of log message:
>
> INFO  [kafka-network-thread-9092-1] [kafka.network.Processor
>   ]: Closing socket connection to /some_ip_that_I_know.
>
> I see at least one every 4-5  seconds. Something I identified was that the
> ip of the closed clients was always from one class of application. This
> application mostly consumes kafka partitions (rarely produces) and it uses
> the SimpleConsumer to consume data, make requests about partition
> leadership (to keep it up to date) and to also make requests about the
> latest offset in a set of partitions (for metrics).
>
> The data consumption happens with long running consumers, but the one off
> requests like partition leadership requests and offset queries happen with
> short-lived consumers that are closed after a request is served. However
> the volume of closing "Closing socket connection" log messages is higher
> than what I think the rate of these short lived requests should be. My
> guess is that something else is going on.
>
> Is there a way for me to track what client is exactly having its connection
> closed? Each of my SimpleConsumer clients uses a particular client name
> when connecting to the broker. Is there a way to enable additional logging
> which would give me this data along with the "Closing socket connection"
> message?
>
> Thanks,
>
> Rajiv
>

Re: Question on Kafka Replication

[Todd Palino]

I put an answer to this on Stack Overflow. Basically, that's not how RF
works for Kafka. It's not a guarantee, it's just how the partitions are
created, and how it is reported when something is down (under replicated
partitions). While there is an option to do auto leader rebalancing,
there's no equivalent option for auto partition migration. It's a fairly
resource intensive task, moving all that data around. If you want to move
replicas around when something is down, you have to do it manually.

That said, it may be interesting to consider.

-Todd


On Tue, Sep 15, 2015 at 7:47 AM, Dhyan  wrote:

> Hi All,
>
> Below is my partition information for the topic **xx_json_topic** .This is
> a Kafka cluster with three nodes .
>
> All nodes up :
>
> Topic: xx_json_topicPartitionCount:4
> ReplicationFactor:2Configs:
> Topic: xx_json_topicPartition: 0Leader: 1   Replicas:
> 3,1   Isr: 3,1
> Topic: xx_json_topicPartition: 1Leader: 2   Replicas:
> 1,2   Isr: 2,1
> Topic: xx_json_topicPartition: 2Leader: 2   Replicas:
> 2,3   Isr: 2,3
> Topic: xx_json_topicPartition: 3Leader: 3   Replicas:
> 3,2   Isr: 2,3
>
> At this point..  if i bring down the node "node-1" ..It looks like below :
>
> Topic: xx_json_topicPartitionCount:4
> ReplicationFactor:2Configs:
> Topic: xx_json_topicPartition: 0Leader: 3   Replicas:
> 3,1   Isr: 3
> Topic: xx_json_topicPartition: 1Leader: 2   Replicas:
> 1,2   Isr: 2
> Topic: xx_json_topicPartition: 2Leader: 2   Replicas:
> 2,3   Isr: 2,3
> Topic: xx_json_topicPartition: 3Leader: 3   Replicas:
> 3,2   Isr: 2,3
>
> My question is ..if kafka knows that the node-1 is down and it needs to
> maintain the replication factor-2 ,wouldn't it make node 3 a replica for
> partition-1 && node-2 a replica for partition-0  then make node-3 and
> node-2 part of their Isr ?
>
> Or you think Kafka doesn't promise that...
> If replication factor is 2 ..It doesn't mean that data will be available
> in atleast 2 nodes at all time(---like consistency level in Cassandra) .
>
> I also have this question posted below :
>
> http://stackoverflow.com/questions/32588784/would-kafka-create-a-new-follower-if-one-of-its-replica-is-down-to-keep-up-with
> <
> http://stackoverflow.com/questions/32588784/would-kafka-create-a-new-follower-if-one-of-its-replica-is-down-to-keep-up-with
> >
>
> —Dhyan

Re: Amount of partitions

[Todd Palino]

How large are your messages compressed? 50k requests/sec could equate to as
little as 50 KB/sec of traffic per topic, 50 GB/sec, or more. The size of
the messages is going to be pretty important when considering overall
throughput here. Additionally, what kind of network interfaces are you
using on your brokers?

If you're working with 400 partitions per topic, that's around 3600
partitions total, or about 2400 partitions per broker (assuming replication
factor 2). This isn't an unreasonable number of partitions per-broker. We
regularly run twice as many partitions as this with only 64 GB of memory. I
don't think partition count is your problem here. Depending on the message
size, you could easily be doing too much.

Let's assume a message size of 2 KB. Given that you have 3 topics doing 50k
req/sec, that would be 100 MB/sec per topic. With 3 brokers, I'm going to
assume that you're running replication factor 2. This means that each
broker has an inbound traffic rate equivalent to 2 of those topics, or 200
MB/sec. Already, this means that you had better be running something more
than a gigabit network interface. If not, I can pin down your first problem
right there. Given that you're seeing a lot of replication messages in the
logs, I would start here by adding more brokers and balancing out your
traffic further.

Each of your brokers also has an outbound traffic rate of 100 MB/sec just
for the inter-broker replication traffic. You're running a mirror maker,
which means that you have to double that. So you're doing 200 MB/sec
inbound, and 200 MB/sec outbound from each broker. How many copies of
mirror maker are you running in your consumer group? If you're only running
1 copy of mirror maker, that means that single box has to handle 300 MB/sec
of inbound and outbound traffic, and it also has to decompress and
recompress all of that traffic. My experience has been that a single thread
of mirror maker (1 server, running with 1 consumer thread) will max out
around 12 MB/sec. You're not too far off, especially if you have problems
on the broker side that is reducing the efficiency of your system.

I don't think you're looking at anything specific to the kernel version or
the JDK at this point, I think you may just be underpowered (depending on
the answers to the questions buried in there).

-Todd


On Mon, Sep 7, 2015 at 1:08 AM, Jörg Wagner <joerg.wagn...@1und1.de> wrote:

> Thank you very much for both replies.
>
> @Tao
> Thanks, I am aware of and have read that article. I am asking because my
> experience is completely different :/. Everytime we go beyond 400
> partitions the cluster really starts breaking apart.
>
> @Todd
> Thank you, very informative.
>
> Or Details:
> 3 Brokers: 192GB Ram, 27 Disks for log.dirs, 9 topics and estimated 50k
> requests / second on 3 of the topics, the others are negligible.
> Ordering is not required, messages are not keyed
>
> The 3 main topics are one per DC (3 DCs) and being mirrored to the others.
>
>
> The issue arises when we use over 400 partitions, which I think we require
> due to performance and mirroring. Partitions get out of sync and the log is
> being spammed by replicator messages. At the core, we start having massive
> stability issues.
>
> Additionally, the mirrormaker only gets 2k Messages per *minute* through
> with a stable setup of 81 partitions (for the 3 main topics).
>
> Has anyone experienced this and can give more insight? We have been doing
> testing for weeks, compared configuration and setups, without finding the
> main cause.
> Can this be a Kernel (version/configuration) or Java(7) issue?
>
> Cheers
> Jörg
>
>
>
> On 04.09.2015 20:24, Todd Palino wrote:
>
>> Jun's post is a good start, but I find it's easier to talk in terms of
>> more
>> concrete reasons and guidance for having fewer or more partitions per
>> topic.
>>
>> Start with the number of brokers in the cluster. This is a good baseline
>> for the minimum number of partitions in a topic, as it will assure balance
>> over the cluster. Of course, if you have lots of topics, you can
>> potentially skip past this as you'll end up with balanced load in the
>> aggregate, but I think it's a good practice regardless. As with all other
>> advice here, there are always exceptions. If you really, really, really
>> need to assure ordering of messages, you might be stuck with a single
>> partition for some use cases.
>>
>> In general, you should pick more partitions if a) the topic is very busy,
>> or b) you have more consumers. Looking at the second case first, you
>> always
>> want to have at least as many partitions in a topic as you have individual
>> consumers in a consumer group. So if you have 16 consumers in a single
>> group, you will want the topi

Re: Amount of partitions

[Todd Palino]

Jun's post is a good start, but I find it's easier to talk in terms of more
concrete reasons and guidance for having fewer or more partitions per topic.

Start with the number of brokers in the cluster. This is a good baseline
for the minimum number of partitions in a topic, as it will assure balance
over the cluster. Of course, if you have lots of topics, you can
potentially skip past this as you'll end up with balanced load in the
aggregate, but I think it's a good practice regardless. As with all other
advice here, there are always exceptions. If you really, really, really
need to assure ordering of messages, you might be stuck with a single
partition for some use cases.

In general, you should pick more partitions if a) the topic is very busy,
or b) you have more consumers. Looking at the second case first, you always
want to have at least as many partitions in a topic as you have individual
consumers in a consumer group. So if you have 16 consumers in a single
group, you will want the topic they consume to have at least 16 partitions.
In fact, you may also want to always have a multiple of the number of
consumers so that you have even distribution. How many consumers you have
in a group is going to be driven more by what you do with the messages once
they are consumed, so here you'll be looking from the bottom of your stack
up, until you get to Kafka.

How busy the topic is is looking from the top down, through the producer,
to Kafka. It's also a little more difficult to provide guidance on. We have
a policy of expanding partitions for a topic whenever the size of the
partition on disk (full retention over 4 days) is larger than 50 GB. We
find that this gives us a few benefits. One is that it takes a reasonable
amount of time when we need to move a partition from one broker to another.
Another is that when we have partitions that are larger than this, the rate
tends to cause problems with consumers. For example, we see mirror maker
perform much better, and have less spiky lag problems, when we stay under
this limit. We're even considering revising the limit down a little, as
we've had some reports from other wildcard consumers that they've had
problems keeping up with topics that have partitions larger than about 30
GB.

The last thing to look at is whether or not you are producing keyed
messages to the topic. If you're working with unkeyed messages, there is no
problem. You can usually add partitions whenever you want to down the road
with little coordination with producers and consumers. If you are producing
keyed messages, there is a good chance you do not want to change the
distribution of keys to partitions at various points in the future when you
need to size up. This means that when you first create the topic, you
probably want to create it with enough partitions to deal with growth over
time, both on the produce and consume side, even if that is too many
partitions right now by other measures. For example, we have one client who
requested 720 partitions for a particular set of topics. The reasoning was
that they are producing keyed messages, they wanted to account for growth,
and they wanted even distribution of the partitions to consumers as they
grow. 720 happens to have a lot of factors, so it was a good number for
them to pick.

As a note, we have up to 5000 partitions per broker right now on current
hardware, and we're moving to new hardware (more disk, 256 GB of memory,
10gig interfaces) where we're going to have up to 12,000. Our default
partition count for most clusters is 8, and we've got topics up to 512
partitions in some places just taking into account the produce rate alone
(not counting those 720-partition topics that aren't that busy). Many of
our brokers run with over 10k open file handles for regular files alone,
and over 50k open when you include network.

-Todd



On Fri, Sep 4, 2015 at 8:11 AM, tao xiao  wrote:

> Here is a good doc to describe how to choose the right number of partitions
>
>
> http://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster/
>
> On Fri, Sep 4, 2015 at 10:08 PM, Jörg Wagner 
> wrote:
>
> > Hello!
> >
> > Regarding the recommended amount of partitions I am a bit confused.
> > Basically I got the impression that it's better to have lots of
> partitions
> > (see information from linkedin etc). On the other hand, a lot of
> > performance benchmarks floating around show only a few partitions are
> being
> > used.
> >
> > Especially when considering the difference between hdd and ssds and also
> > the amount thereof, what is the way to go?
> >
> > In my case, I seem to have the best stability and performance issues with
> > few partitions *per hdd*, and only one io thread per disk.
> >
> > What are your experiences and recommendations?
> >
> > Cheers
> > Jörg
> >
>
>
>
> --
> Regards,
> Tao
>

Re: How to monitor lag when "kafka" is used as offset.storage?

[Todd Palino]

You can use the emailer config in Burrow to send alerts directly (it will
monitor specific groups and send emails out when there is a problem). If
you need something more complex than that, I think the best practice is
always to send the output into an general alert/notification system.

-Todd

On Wednesday, September 2, 2015, shahab  wrote:

> Thanks Noah. I installed Burrow and played with it a little bit. It seems
> as you pointed out I need to implement the alerting system myself. Do you
> know any other Kafka tools that can give alerts?
>
> best,
> /Shahab
>
> On Wed, Sep 2, 2015 at 1:44 PM, noah >
> wrote:
>
> > We use Burrow . There are rest
> > endpoints you can use to get offsets and manually calculate lag, but if
> you
> > are focused on alerting, I'd use it's consumer statuses as they are a bit
> > smarter than a simple lag calculation.
> >
> > On Wed, Sep 2, 2015 at 4:08 AM shahab  > wrote:
> >
> > > Hi,
> > >
> > > I wonder how we can monitor lag (difference between consumer offset and
> > log
> > > ) when "kafka" is set as offset.storage?  because the
> "kafka-run-class.sh
> > > kafka.tools.ConsumerOffsetChecker ... " does work only when zookeeper
> is
> > > used as storage manager.
> > >
> > > best,
> > > /Shahab
> > >
> >
>

Re: API to query cluster metadata on-demand

[Todd Palino]

What Gwen said :)

We developed a python web service internally called Menagerie that provides
this functionality for both Kafka and Zookeeper. We use it to drive a web
dashboard for stats, our (old style) lag checking, and some other CLI
tools. Unfortunately it ties into too much internal LinkedIn tooling for us
to open source.

That's one of the reasons we released Burrow (
https://github.com/linkedin/Burrow). The primary use is to do lag checking
for consumers as a service. But I'm also moving functionality from
Menagerie into it. Right now you can use it to fetch topic lists, partition
counts, and broker offsets. You can also get information for consumers (as
long as they are committing offsets to Kafka and not ZK).

If it looks useful and there's some bit of info you'd like it to provide,
you can submit a github issue and I'll take a look at it.

-Todd

On Thursday, September 3, 2015, Andrew Otto  wrote:

> If you don’t mind doing it with a C CLI:
>
> https://github.com/edenhill/kafkacat
>
> $ kafkacat -L -b mybroker
>
> But, uhhh, you probably want a something in the Java API.
>
> :)
>
>
> > On Sep 3, 2015, at 13:58, Gwen Shapira >
> wrote:
> >
> > Ah, I wish.
> >
> > We are working on it :)
> >
> > On Thu, Sep 3, 2015 at 9:10 AM, Simon Cooper <
> > simon.coo...@featurespace.co.uk > wrote:
> >
> >> Is there a basic interface in the new client APIs to get the list of
> >> topics on a cluster, and get information on the topics (offsets, sizes,
> >> etc), without having to deal with a producer or consumer? I just want a
> >> basic synchronous API to query the metadata as-is. Does this exist in
> some
> >> form?
> >>
> >> Thanks,
> >> Simon
> >>
>
>

Re: Recovery of Kafka cluster takes very long time

[Todd Palino]

It looks like you did an unclean shutdown of the cluster, in which case
each open log segment in each partition needs to be checked upon startup.
It doesn't really have anything to do with RF=3 specifically, but it does
mean that each of your brokers has 6000 partitions to check.

What is the setting of recovery.threads.per.data.dir in your broker
configuration? The default is 1, which means that upon startup and
shutdown, the broker only uses 1 thread for checking/closing log segments.
If you increase this, it will parallelize both the startup and shutdown
process. This is particularly helpful for recovering from unclean shutdown.
We generally set it to the number of CPUs in the system, because we want a
fast recovery.

-Todd


On Mon, Aug 10, 2015 at 8:57 AM, Alexey Sverdelov 
alexey.sverde...@googlemail.com wrote:

 Hi all,

 I have a 3 node Kafka cluster. There are ten topics, every topic has 600
 partitions with RF3.

 So, after cluster restart I can see the following log message like INFO
 Recovering unflushed segment 0 in log... and the complete recovery of 3
 nodes takes about 2+ hours.

 I don't know why it takes so long? Is it because of RF=3?

 Have a nice day,
 Alexey

Re: Specify leader when doing partition reassignment

[Todd Palino]

To make sure you have a complete answer here, the order of the replica list
that you specify in the partition reassignment will affect the leader
selection, but if the current leader is in the new replica list, it will
not change the leadership to change.

That is, if your current replica list is [1, 2] and the leader is 1, and
you change the replica list to [3, 1] nothing happens immediately except
that 2 is removed as a replica and 3 is added. Upon a failure of broker 1
or a preferred replica election, broker 3 will take over leadership. If the
new replica list is [3, 4], then upon completion of the reassignment,
broker 3 will be selected as the new leader.

The controller will prefer to select the first broker in the replica list
as the leader. This means that if you do a partition reassignment and
change the replica list from [1, 2] to [2, 1], nothing happens at first.
But upon the next preferred replica election, broker 2 will be selected as
the leader.

-Todd


On Wed, Aug 5, 2015 at 3:37 AM, tao xiao xiaotao...@gmail.com wrote:

 I think I figured it out. As pointed out in

 https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-6.ReassignPartitionsTool

 If the leader is not in the reassigned replica a new leader will be elected
 and the tool doesn't pick the first one from the RAR as the leader

 On Wed, 5 Aug 2015 at 18:32 Jilin Xie jilinxie1...@gmail.com wrote:

  Check the --replica-assignment  parameter of the kafka-topics.sh.
  It does what you need.
  And there should also be similar configs in the api if you wanna do so by
  coding.
 
  On Wed, Aug 5, 2015 at 6:18 PM, tao xiao xiaotao...@gmail.com wrote:
 
   Hi team,
  
   Is it possible to specify a leader broker for each topic partition when
   doing partition reassignment?
  
   For example I have following json. Is the first broker in the replicas
  list
   by default the leader of the partition e.g. broker 3 is the leader of
  topic
   test5 and broker 2 is the leader of topic test3. or does Kafka
   automatically pick the leader based on the leadership ratio
   if auto.leader.rebalance.enable is on
  
   {
 version: 1,
 partitions: [
   {
 topic: test5,
 partition: 0,
 replicas: [3, 1]
   },
   {
 topic: test3,
 partition: 0,
 replicas: [2,3]
   },
   {
 topic: test1,
 partition: 0,
 replicas: [1,2]
   }
 ]
   }
  
 

Re: Documentation typo for offsets.topic.replication.factor ?

[Todd Palino]

That's exactly right. We've been talking about this internally at LinkedIn, and 
how to solve it. I think the best option would be to have the broker throw an 
error on offset commits until there are enough brokers to fulfill the 
configured RF.

We've seen this several times now when bootstrapping test and development 
clusters. You can work around it by making sure all consumers are shut down 
until you have all the brokers started. But that's not a good long term 
solution for this problem. 

-Todd


 On Aug 5, 2015, at 6:28 PM, James Cheng jch...@tivo.com wrote:
 
 Hi,
 
 My kafka cluster has a __consumer_offsets topic with 50 partitions (the 
 default for offsets.topic.num.partitions) but with a replication factor of 
 just 1 (the default for offsets.topic.replication.factor should be 3).
 
 From the docs http://kafka.apache.org/documentation.html:
 
 offsets.topic.replication.factor3The replication factor for the 
 offset commit topic. A higher setting (e.g., three or four) is recommended in 
 order to ensure higher availability. If the offsets topic is created when 
 fewer brokers than the replication factor then the offsets topic will be 
 created with fewer replicas.
 
 
 I'm guessing there's a typo there? I'm guessing it should be:
 
 If the offsets topic is created when fewer brokers than the replication 
 factor [are active], then the offsets topic will be created with fewer 
 replicas.
 
 Or something along those lines?
 
 Thanks,
 -James
 
 

Re: Custom Zookeeper install with kafka

[Todd Palino]

Yes, we use ZK 3.4.6 exclusively at LinkedIn and there's no problem.

-Todd

 On Jul 22, 2015, at 9:49 AM, Adam Dubiel dubiel.a...@gmail.com wrote:
 
 Hi,
 
 I don't think it matters much which version of ZK will you use (meaning
 minor/patch versions). We have been using 3.4.6 for some time and it works
 flawlessly.
 
 BR,
 Adam
 
 2015-07-22 18:40 GMT+02:00 Prabhjot Bharaj prabhbha...@gmail.com:
 
 Hi,
 
 I've read on the Kafka documentation page that the zookeeper version used
 is 3.3.4
 
 However, at my work, I've noticed certain problems with v3.3.4 (and the
 problems are documented here: http://zookeeper.apache.org/releases.html
 
 The latest stable version of zookeeper is 3.4.6 and has a lot of bug fixes
 from 3.3.4
 
 Has anyone tried using this version of zookeeper with kafka 0.8.2.1 ?
 
 Regards,
 Prabhjot
 

Re: stunning error - Request of length 1550939497 is not valid, it is larger than the maximum size of 104857600 bytes

[Todd Palino]

This is interesting. We have seen something similar internally at LinkedIn
with one particular topic (and Avro schema), and only once in a while.
We've seen it happen 2 or 3 times so far. We had chalked it up to bad
content in the message, figuring that the sender was doing something like
sending a long stream of a single character, in error, which was creating a
highly compressible message. Given these cases, I'm no longer certain
that's the case.

Becket, you had been taking a look at this internally. Do you have any
thoughts on this?

-Todd


On Tue, Jul 14, 2015 at 11:18 AM, JIEFU GONG jg...@berkeley.edu wrote:

 @Gwen
 I am having a very very similar issue where I am attempting to send a
 rather small message and it's blowing up on me (my specific error is:
 Invalid receive (size = 1347375956 larger than 104857600)). I tried to
 change the relevant settings but it seems that this particular request is
 of 1340 mbs (and davids will be 1500 mb) and attempting to change the
 setting will give you another error saying there is not enough memory in
 the java heap. Any insight here?

 Specifically I am speculating the issue is indeed what Shayne has said
 about encoding: I am trying to use apachebench to send a post request to a
 kafka server but it is returning the above error -- do I have to format the
 data in any way as this might be the reason why I'm experience this issue.


 On Sun, Jul 12, 2015 at 6:35 AM, Shayne S shaynest...@gmail.com wrote:

  Your payload is so small that I suspect it's an encoding issue. Is your
  producer set to expect a byte array and you're passing a string? Or vice
  versa?
 
  On Sat, Jul 11, 2015 at 11:08 PM, David Montgomery 
  davidmontgom...@gmail.com wrote:
 
   I cant send this s simple payload using python.
  
   topic: topic-test-development
   payload: {utcdt: 2015-07-12T03:59:36, ghznezzhmx: apple}
  
  
   No handlers could be found for logger kafka.conn
   Traceback (most recent call last):
 File /home/ubuntu/workspace/feed-tests/tests/druid-adstar.py, line
  81,
   in module
   test_send_data_to_realtimenode()
 File /home/ubuntu/workspace/feed-tests/tests/druid-adstar.py, line
  38,
   in test_send_data_to_realtimenode
   response = producer.send_messages(test_topic,test_payload)
 File
 /usr/local/lib/python2.7/dist-packages/kafka/producer/simple.py,
   line 54, in send_messages
   topic, partition, *msg
 File /usr/local/lib/python2.7/dist-packages/kafka/producer/base.py,
   line 349, in send_messages
   return self._send_messages(topic, partition, *msg)
 File /usr/local/lib/python2.7/dist-packages/kafka/producer/base.py,
   line 390, in _send_messages
   fail_on_error=self.sync_fail_on_error
 File /usr/local/lib/python2.7/dist-packages/kafka/client.py, line
  480,
   in send_produce_request
   (not fail_on_error or not self._raise_on_response_error(resp))]
 File /usr/local/lib/python2.7/dist-packages/kafka/client.py, line
  247,
   in _raise_on_response_error
   raise resp
   kafka.common.FailedPayloadsError
  
   Here is what is in my logs
   [2015-07-12 03:29:58,103] INFO Closing socket connection to
   /xxx.xxx.xxx.xxx due to invalid request: Request of length 1550939497
 is
   not valid, it is larger than the maximum size of 104857600 bytes.
   (kafka.network.Processor)
  
  
  
   Server is 4 gigs of ram.
  
   I used export KAFKA_HEAP_OPTS=-Xmx256M -Xms128M in
 kafka-server-start.sh
  
   So.why?
  
 



 --

 Jiefu Gong
 University of California, Berkeley | Class of 2017
 B.A Computer Science | College of Letters and Sciences

 jg...@berkeley.edu elise...@berkeley.edu | (925) 400-3427

Re: Leap Second Troubles

[Todd Palino]

OK, in that case then I'm thinking that you ran into issues that were a
natural result of the Zookeeper ensemble having very high CPU usage.
Unfortunate, but this would not be an unexpected situation when your ZK
ensemble is having significant problems.

-Todd


On Fri, Jul 10, 2015 at 10:21 AM, Christofer Hedbrandh 
christo...@knewton.com wrote:

 Todd, the Kafka problems started when one of three ZooKeeper nodes was
 restarted.

 On Thu, Jul 9, 2015 at 12:10 PM, Todd Palino tpal...@gmail.com wrote:

  Did you hit the problems in the Kafka brokers and consumers during the
  Zookeeper problem, or after you had already cleared it?
 
  For us, we decided to skip the leap second problem (even though we're
  supposedly on a version that doesn't have that bug) by shutting down ntpd
  everywhere and then allowing it to slowly adjust the time afterwards
  without sending the leap second.
 
  -Todd
 
 
  On Thu, Jul 9, 2015 at 7:58 AM, Christofer Hedbrandh 
  christo...@knewton.com
   wrote:
 
   Hi Kafka users,
  
   ZooKeeper in our staging environment was running on a very old ubuntu
   version, that was exposed to the leap second causes spuriously high
 CPU
   usage bug.
  
   https://bugs.launchpad.net/ubuntu/+source/linux/+bug/1020285
  
   As a result, when the leap second arrived, the ZooKeeper CPU usage went
  up
   to 100% and stayed there. In response to this, we restarted one
 ZooKeeper
   process. The ZooKeeper restart unfortunately made the situation much
  worse
   as we hit three different (possibly related) Kafka problems. We are
 using
   Kafka 0.8.2 brokers, consumers and producers.
  
  
   #1
   One of our three brokers was kicked out or ISR for some (most but not
  all)
   partitions, and was continuously logging Cached zkVersion [XX] not
 equal
   to that in zookeeper, skip updating ISR over and over (until I
  eventually
   stopped this broker).
  
   INFO Partition [topic-x,xx] on broker 1: Shrinking ISR for partition
   [topic-x,xx] from 1,2,3 to 1 (kafka.cluster.Partition)
   INFO Partition [topic-x,xx] on broker 1: Cached zkVersion [62] not
 equal
  to
   that in zookeeper, skip updating ISR (kafka.cluster.Partition)
   INFO Partition [topic-y,yy] on broker 1: Shrinking ISR for partition
   [topic-y,yy] from 1,2,3 to 1 (kafka.cluster.Partition)
   INFO Partition [topic-y,yy] on broker 1: Cached zkVersion [39] not
 equal
  to
   that in zookeeper, skip updating ISR (kafka.cluster.Partition)
   INFO Partition [topic-z,zz] on broker 1: Shrinking ISR for partition
   [topic-z,zz] from 1,2,3 to 1 (kafka.cluster.Partition)
   INFO Partition [topic-z,zz] on broker 1: Cached zkVersion [45] not
 equal
  to
   that in zookeeper, skip updating ISR (kafka.cluster.Partition)
   etc.
  
   Searching the users@kafka.apache.org archive and Googling for this log
   output, gives me similar descriptions but nothing that exactly
 describes
   this.
   It is very similar to this, but without the ERROR Conditional update
 of
   path ... part.
   https://www.mail-archive.com/users@kafka.apache.org/msg07044.html
  
  
   #2
   The remaining two brokers were logging this every five seconds or so.
   INFO conflict in /brokers/ids/xxx data:
  
  
 
 {jmx_port:xxx,timestamp:1435712198759,host:xxx,version:1,port:9092}
   stored data:
  
  
 
 {jmx_port:xxx,timestamp:1435711782536,host:xxx,version:1,port:9092}
   (kafka.utils.ZkUtils$)
   INFO I wrote this conflicted ephemeral node
  
  
 
 [{jmx_port:xxx,timestamp:1435712198759,host:xxx,version:1,port:9092}]
   at /brokers/ids/xxx a while back in a different session, hence I will
   backoff for this node to be deleted by Zookeeper and retry
   (kafka.utils.ZkUtils$)
  
   It sounds very much like we hit this bug
   https://issues.apache.org/jira/browse/KAFKA-1387
  
  
   #3
   The most serious issue that resulted was that some consumer groups
 failed
   to claim all partitions. When using the ConsumerOffsetChecker, the
 owner
  of
   some partitions was listed as none, the lag was constantly
 increasing,
   and it was clear that no consumers were processing these messages.
  
   It is exactly what Dave Hamilton is describing here, but from this
 email
   chain no one seems to know what caused it.
   https://www.mail-archive.com/users%40kafka.apache.org/msg13364.html
  
   It may be reasonable to assume that the consumer rebalance failures we
  also
   saw has something to do with this. But why the rebalance failed is
 still
   unclear.
  
   ERROR k.c.ZookeeperConsumerConnector: error during syncedRebalance
   kafka.common.ConsumerRebalanceFailedException: xxx can't rebalance
 after
  4
   retries
   at
  
  
 
 kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:633)
   at
  
  
 
 kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:551)
  
  
   I am curious to hear if anyone else had similar problems to this?
  
   And also if anyone can say

Re: Leap Second Troubles

[Todd Palino]

Did you hit the problems in the Kafka brokers and consumers during the
Zookeeper problem, or after you had already cleared it?

For us, we decided to skip the leap second problem (even though we're
supposedly on a version that doesn't have that bug) by shutting down ntpd
everywhere and then allowing it to slowly adjust the time afterwards
without sending the leap second.

-Todd


On Thu, Jul 9, 2015 at 7:58 AM, Christofer Hedbrandh christo...@knewton.com
 wrote:

 Hi Kafka users,

 ZooKeeper in our staging environment was running on a very old ubuntu
 version, that was exposed to the leap second causes spuriously high CPU
 usage bug.

 https://bugs.launchpad.net/ubuntu/+source/linux/+bug/1020285

 As a result, when the leap second arrived, the ZooKeeper CPU usage went up
 to 100% and stayed there. In response to this, we restarted one ZooKeeper
 process. The ZooKeeper restart unfortunately made the situation much worse
 as we hit three different (possibly related) Kafka problems. We are using
 Kafka 0.8.2 brokers, consumers and producers.


 #1
 One of our three brokers was kicked out or ISR for some (most but not all)
 partitions, and was continuously logging Cached zkVersion [XX] not equal
 to that in zookeeper, skip updating ISR over and over (until I eventually
 stopped this broker).

 INFO Partition [topic-x,xx] on broker 1: Shrinking ISR for partition
 [topic-x,xx] from 1,2,3 to 1 (kafka.cluster.Partition)
 INFO Partition [topic-x,xx] on broker 1: Cached zkVersion [62] not equal to
 that in zookeeper, skip updating ISR (kafka.cluster.Partition)
 INFO Partition [topic-y,yy] on broker 1: Shrinking ISR for partition
 [topic-y,yy] from 1,2,3 to 1 (kafka.cluster.Partition)
 INFO Partition [topic-y,yy] on broker 1: Cached zkVersion [39] not equal to
 that in zookeeper, skip updating ISR (kafka.cluster.Partition)
 INFO Partition [topic-z,zz] on broker 1: Shrinking ISR for partition
 [topic-z,zz] from 1,2,3 to 1 (kafka.cluster.Partition)
 INFO Partition [topic-z,zz] on broker 1: Cached zkVersion [45] not equal to
 that in zookeeper, skip updating ISR (kafka.cluster.Partition)
 etc.

 Searching the users@kafka.apache.org archive and Googling for this log
 output, gives me similar descriptions but nothing that exactly describes
 this.
 It is very similar to this, but without the ERROR Conditional update of
 path ... part.
 https://www.mail-archive.com/users@kafka.apache.org/msg07044.html


 #2
 The remaining two brokers were logging this every five seconds or so.
 INFO conflict in /brokers/ids/xxx data:

 {jmx_port:xxx,timestamp:1435712198759,host:xxx,version:1,port:9092}
 stored data:

 {jmx_port:xxx,timestamp:1435711782536,host:xxx,version:1,port:9092}
 (kafka.utils.ZkUtils$)
 INFO I wrote this conflicted ephemeral node

 [{jmx_port:xxx,timestamp:1435712198759,host:xxx,version:1,port:9092}]
 at /brokers/ids/xxx a while back in a different session, hence I will
 backoff for this node to be deleted by Zookeeper and retry
 (kafka.utils.ZkUtils$)

 It sounds very much like we hit this bug
 https://issues.apache.org/jira/browse/KAFKA-1387


 #3
 The most serious issue that resulted was that some consumer groups failed
 to claim all partitions. When using the ConsumerOffsetChecker, the owner of
 some partitions was listed as none, the lag was constantly increasing,
 and it was clear that no consumers were processing these messages.

 It is exactly what Dave Hamilton is describing here, but from this email
 chain no one seems to know what caused it.
 https://www.mail-archive.com/users%40kafka.apache.org/msg13364.html

 It may be reasonable to assume that the consumer rebalance failures we also
 saw has something to do with this. But why the rebalance failed is still
 unclear.

 ERROR k.c.ZookeeperConsumerConnector: error during syncedRebalance
 kafka.common.ConsumerRebalanceFailedException: xxx can't rebalance after 4
 retries
 at

 kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:633)
 at

 kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:551)


 I am curious to hear if anyone else had similar problems to this?

 And also if anyone can say if these are all known bugs that are being
 tracked with some ticket number?


 Thanks,
 Christofer

 P.S. Eventually after ZooKeeper and Kafka broker and consumer restarts
 everything returned to normal.

Re: Is trunk safe for production?

[Todd Palino]

Yes and no. We're running a version about a month behind trunk at any given
time here at LinkedIn. That's generally the amount of time we spend testing
and going through our release process internally (less if there are no
problems). So it can be done.

That said, we also have several Kafka contributors and a committer who work
on Kafka constantly here. When talking to others about how we run Kafka at
LinkedIn, I usually say we run trunk so you don't have to. Unless you
have someone to track down and work on fixing the bugs, it's probably a
good idea to stick with the release versions, unless you run in a
development environment where you can tolerate failures and performance
regressions.

-Todd


On Tue, Jun 23, 2015 at 2:59 AM, Achanta Vamsi Subhash 
achanta.va...@flipkart.com wrote:

 I am planning to use for the producer part. How stable is trunk generally?

 --
 Regards
 Vamsi Subhash

 --



 --

 This email and any files transmitted with it are confidential and intended
 solely for the use of the individual or entity to whom they are addressed.
 If you have received this email in error please notify the system manager.
 This message contains confidential information and is intended only for the
 individual named. If you are not the named addressee you should not
 disseminate, distribute or copy this e-mail. Please notify the sender
 immediately by e-mail if you have received this e-mail by mistake and
 delete this e-mail from your system. If you are not the intended recipient
 you are notified that disclosing, copying, distributing or taking any
 action in reliance on the contents of this information is strictly
 prohibited. Although Flipkart has taken reasonable precautions to ensure no
 viruses are present in this email, the company cannot accept responsibility
 for any loss or damage arising from the use of this email or attachments