Experiences with corrupted messages

[Jörg Wagner]

Hey everyone,

I've been having some issues with corrupted messages and mirrormaker as 
I wrote previously. Since there was no feedback, I want to ask a new 
question:


Did you ever have corrupted messages in kafka? Did things break? How did 
you recover or work around that?


Thanks
Jörg

Consumer stops reading messages after some time

[Alexey Sverdelov]

Hi,

I'm facing an issue with high level kafka consumer (0.8.2.0) - after
consuming some amount of data one of our consumers stops. After restart it
consumes some messages and stops again with no error/exception or warning.

After some investigation I found that the "ConsumerFetcherThread" for my
main topic gets stuck: it makes no progress and writes no log messages.


Consumer for my second topic seems to be ok - I can see some FetchRequests.

Any thoughts?

Alexey

Re: Consumer stops reading messages after some time

[Alexey Sverdelov]

Answering my own message, the problem with consumer was this exception:

==
ERROR c.u.u.e.impl.kafka.KafkaConsumer  - Error consuming message stream:
 # kafka.message.InvalidMessageException: Message is corrupt (stored crc =
3801080313, computed crc = 2728178222)
 # \x09at kafka.message.Message.ensureValid(Message.scala:166)
~[org.apache.kafka.kafka_2.11-0.8.2.0.jar:na]
==

Any ideas how can I simple ignore such messages at all?

Thanks.

On Thu, Oct 1, 2015 at 1:05 PM, Alexey Sverdelov <
alexey.sverde...@googlemail.com> wrote:

> Hi,
>
> I'm facing an issue with high level kafka consumer (0.8.2.0) - after
> consuming some amount of data one of our consumers stops. After restart it
> consumes some messages and stops again with no error/exception or warning.
>
> After some investigation I found that the "ConsumerFetcherThread" for my
> main topic gets stuck: it makes no progress and writes no log messages.
>
>
> Consumer for my second topic seems to be ok - I can see some FetchRequests.
>
> Any thoughts?
>
> Alexey
>

Re: What happens when ISR is behind leader

[pushkar priyadarshi]

Hi,

There are two properties which determines when does a replica falls off
sync.Look for replica.lag.time.max.ms and replica.lag.max.messages .If a
replica goes out of sync then it would not be even considered for leader
election.

Regards,
Pushkar

On Wed, Sep 30, 2015 at 9:44 AM, Shushant Arora 
wrote:

> Hi
>
> I have a kafka cluster with 2 brokers and replication as 2.
> Now say for a partition P1 leader broker b1 has offsets 1-10 and ISR broker
> is behind leader and now it has data for offsets (1-5) only. Now broker B1
> gets down and kafka elects B2 as leader for partition P1. Now new write for
> partition P1 will happen on B2 - what will be the offset of new message
> will it start from (5+1)=6 or (10+1)=11?
>
> And if it starts from 11 ? will offsets 6-10 will be missing ?
>
> Thanks
>

Re: are 0.8.2.1 and 0.9.0.0 compatible?

[Jason Rosenberg]

Of course, that documentation needs to be updated to refer to '0.9.X'!

Also, I'm wondering if the last step there should be changed to remove the
property altogether and restart (rather than setting it to the new
version), since once the code is updated, it will use that by default?

On Thu, Oct 1, 2015 at 1:48 PM, Grant Henke  wrote:

> Hi Richard,
>
> You are correct that version will now be 0.9.0 and anything referencing
> 0.8.3 is being changed. You are also correct in the there have been wire
> protocol changes that break compatibility. However, backwards compatibility
> exists and you should always upgrade your brokers before upgrading your
> clients in order to avoid issues (In the future KIP-35
>  may change that).
>
> It's also worth noting that if you are performing a rolling upgrade of your
> brokers, you need to be sure brokers running the new protocol know to
> communicate with the old version to remain compatible during the bounce.
> This is done using the inter.broker.protocol.version property. More on that
> topic can be read here:
> https://kafka.apache.org/083/documentation.html#upgrade
>
> Hopefully that helps clear things up.
>
> Thank you,
> Grant
>
>
>
>
>
> On Thu, Oct 1, 2015 at 12:21 PM, Richard Lee  wrote:
>
> > Note the 0.8.3-SNAPSHOT has recently been renamed 0.9.0.0-SNAPSHOT.
> >
> > In any event, the major version number change could indicate that there
> > has, in fact, been some sort of incompatible change.  Using 0.9.0.0, I'm
> > also unable to use the kafka-console-consumer.sh to read from a 0.8.2.1
> > broker, but it works fine with a 0.9.0.0 broker.
> >
> > Some validation from a kafka expert that broker forward compatibility (or
> > client backward compatibility) is not supported would be appreciated, and
> > that this isn't just a case of some sort of local, fixable
> misconfiguration.
> >
> > Thanks!
> > Richard
> >
> > On 09/30/2015 11:17 AM, Doug Tomm wrote:
> >
> >> hello,
> >>
> >> i've got a set of broker nodes running 0.8.2.1.  on my laptop i'm also
> >> running 0.8.2.1, and i have a single broker node and mirrormaker there.
> >> i'm also using kafka-console-consumer.sh on the mac to display messages
> on
> >> a favorite topic being published from the broker nodes.  there are no
> >> messages on the topic, but everything is well-behaved.  i can inject
> >> messages with kafkacat and everything is fine.
> >>
> >> but then!
> >>
> >> on the laptop i switched everything to 0.8.3 but left the broker nodes
> >> alone.  now when i run mirrormaker i see this:
> >>
> >> [2015-09-30 10:44:55,090] WARN
> >>
> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
> >> Error in fetch
> kafka.consumer.ConsumerFetcherThread$FetchRequest@61cb11c5.
> >> Possible cause: java.nio.BufferUnderflowException
> >> (kafka.consumer.ConsumerFetcherThread)
> >> [2015-09-30 10:44:55,624] WARN
> >>
> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
> >> Error in fetch
> kafka.consumer.ConsumerFetcherThread$FetchRequest@3c7bb986.
> >> Possible cause: java.nio.BufferUnderflowException
> >> (kafka.consumer.ConsumerFetcherThread)
> >> [2015-09-30 10:44:56,181] WARN
> >>
> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
> >> Error in fetch
> kafka.consumer.ConsumerFetcherThread$FetchRequest@1d4fbd2c.
> >> Possible cause: java.nio.BufferUnderflowException
> >> (kafka.consumer.ConsumerFetcherThread)
> >> [2015-09-30 10:44:56,726] WARN
> >>
> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
> >> Error in fetch
> kafka.consumer.ConsumerFetcherThread$FetchRequest@59e67b2f.
> >> Possible cause: java.nio.BufferUnderflowException
> >> (kafka.consumer.ConsumerFetcherThread)
> >>
> >> if i use kafkacat to generate a message on the topic i see
> >> IllegalArgumentExceptions instead.
> >>
> >> this suggests that the two versions of kafka aren't compatible. is this
> >> the case?  does the whole ecosystem need to be on the same version?
> >>
> >> thank you,
> >> doug
> >>
> >>
> >
>
>
> --
> Grant Henke
> Software Engineer | Cloudera
> gr...@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>

Re: are 0.8.2.1 and 0.9.0.0 compatible?

[Grant Henke]

Hi Richard,

You are correct that version will now be 0.9.0 and anything referencing
0.8.3 is being changed. You are also correct in the there have been wire
protocol changes that break compatibility. However, backwards compatibility
exists and you should always upgrade your brokers before upgrading your
clients in order to avoid issues (In the future KIP-35
 may change that).

It's also worth noting that if you are performing a rolling upgrade of your
brokers, you need to be sure brokers running the new protocol know to
communicate with the old version to remain compatible during the bounce.
This is done using the inter.broker.protocol.version property. More on that
topic can be read here:
https://kafka.apache.org/083/documentation.html#upgrade

Hopefully that helps clear things up.

Thank you,
Grant





On Thu, Oct 1, 2015 at 12:21 PM, Richard Lee  wrote:

> Note the 0.8.3-SNAPSHOT has recently been renamed 0.9.0.0-SNAPSHOT.
>
> In any event, the major version number change could indicate that there
> has, in fact, been some sort of incompatible change.  Using 0.9.0.0, I'm
> also unable to use the kafka-console-consumer.sh to read from a 0.8.2.1
> broker, but it works fine with a 0.9.0.0 broker.
>
> Some validation from a kafka expert that broker forward compatibility (or
> client backward compatibility) is not supported would be appreciated, and
> that this isn't just a case of some sort of local, fixable misconfiguration.
>
> Thanks!
> Richard
>
> On 09/30/2015 11:17 AM, Doug Tomm wrote:
>
>> hello,
>>
>> i've got a set of broker nodes running 0.8.2.1.  on my laptop i'm also
>> running 0.8.2.1, and i have a single broker node and mirrormaker there.
>> i'm also using kafka-console-consumer.sh on the mac to display messages on
>> a favorite topic being published from the broker nodes.  there are no
>> messages on the topic, but everything is well-behaved.  i can inject
>> messages with kafkacat and everything is fine.
>>
>> but then!
>>
>> on the laptop i switched everything to 0.8.3 but left the broker nodes
>> alone.  now when i run mirrormaker i see this:
>>
>> [2015-09-30 10:44:55,090] WARN
>> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
>> Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@61cb11c5.
>> Possible cause: java.nio.BufferUnderflowException
>> (kafka.consumer.ConsumerFetcherThread)
>> [2015-09-30 10:44:55,624] WARN
>> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
>> Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@3c7bb986.
>> Possible cause: java.nio.BufferUnderflowException
>> (kafka.consumer.ConsumerFetcherThread)
>> [2015-09-30 10:44:56,181] WARN
>> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
>> Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@1d4fbd2c.
>> Possible cause: java.nio.BufferUnderflowException
>> (kafka.consumer.ConsumerFetcherThread)
>> [2015-09-30 10:44:56,726] WARN
>> [ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
>> Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@59e67b2f.
>> Possible cause: java.nio.BufferUnderflowException
>> (kafka.consumer.ConsumerFetcherThread)
>>
>> if i use kafkacat to generate a message on the topic i see
>> IllegalArgumentExceptions instead.
>>
>> this suggests that the two versions of kafka aren't compatible. is this
>> the case?  does the whole ecosystem need to be on the same version?
>>
>> thank you,
>> doug
>>
>>
>


-- 
Grant Henke
Software Engineer | Cloudera
gr...@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke

Re: are 0.8.2.1 and 0.9.0.0 compatible?

[Richard Lee]

Great.. that makes sense.  Forward compatibility by brokers is likely 
hard, tho it would be nice if clients were backward compatible.  I 
guess, tho, implementing that requires KIP-35.


Thanks for the 0.9.0.0 rolling update pointer.

Richard

On 10/01/2015 10:48 AM, Grant Henke wrote:

Hi Richard,

You are correct that version will now be 0.9.0 and anything referencing
0.8.3 is being changed. You are also correct in the there have been wire
protocol changes that break compatibility. However, backwards compatibility
exists and you should always upgrade your brokers before upgrading your
clients in order to avoid issues (In the future KIP-35
 may change that).

It's also worth noting that if you are performing a rolling upgrade of your
brokers, you need to be sure brokers running the new protocol know to
communicate with the old version to remain compatible during the bounce.
This is done using the inter.broker.protocol.version property. More on that
topic can be read here:
https://kafka.apache.org/083/documentation.html#upgrade

Hopefully that helps clear things up.

Thank you,
Grant





On Thu, Oct 1, 2015 at 12:21 PM, Richard Lee  wrote:


Note the 0.8.3-SNAPSHOT has recently been renamed 0.9.0.0-SNAPSHOT.

In any event, the major version number change could indicate that there
has, in fact, been some sort of incompatible change.  Using 0.9.0.0, I'm
also unable to use the kafka-console-consumer.sh to read from a 0.8.2.1
broker, but it works fine with a 0.9.0.0 broker.

Some validation from a kafka expert that broker forward compatibility (or
client backward compatibility) is not supported would be appreciated, and
that this isn't just a case of some sort of local, fixable misconfiguration.

Thanks!
Richard

On 09/30/2015 11:17 AM, Doug Tomm wrote:


hello,

i've got a set of broker nodes running 0.8.2.1.  on my laptop i'm also
running 0.8.2.1, and i have a single broker node and mirrormaker there.
i'm also using kafka-console-consumer.sh on the mac to display messages on
a favorite topic being published from the broker nodes.  there are no
messages on the topic, but everything is well-behaved.  i can inject
messages with kafkacat and everything is fine.

but then!

on the laptop i switched everything to 0.8.3 but left the broker nodes
alone.  now when i run mirrormaker i see this:

[2015-09-30 10:44:55,090] WARN
[ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@61cb11c5.
Possible cause: java.nio.BufferUnderflowException
(kafka.consumer.ConsumerFetcherThread)
[2015-09-30 10:44:55,624] WARN
[ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@3c7bb986.
Possible cause: java.nio.BufferUnderflowException
(kafka.consumer.ConsumerFetcherThread)
[2015-09-30 10:44:56,181] WARN
[ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@1d4fbd2c.
Possible cause: java.nio.BufferUnderflowException
(kafka.consumer.ConsumerFetcherThread)
[2015-09-30 10:44:56,726] WARN
[ConsumerFetcherThread-tivo_kafka_110339-mbpr.local-1443635093396-c55cbafb-0-5],
Error in fetch kafka.consumer.ConsumerFetcherThread$FetchRequest@59e67b2f.
Possible cause: java.nio.BufferUnderflowException
(kafka.consumer.ConsumerFetcherThread)

if i use kafkacat to generate a message on the topic i see
IllegalArgumentExceptions instead.

this suggests that the two versions of kafka aren't compatible. is this
the case?  does the whole ecosystem need to be on the same version?

thank you,
doug

Re: Experiences with corrupted messages

[Lance Laursen]

Hey Jörg,

Unfortunately when the high level consumer hits a corrupt message, it
enters an invalid state and closes. The only way around this is to iterate
your offset by 1 in order to skip the corrupt message. This is currently
not automated. You can catch this exception if you are using the simple
consumer client, but unfortunately mirrormaker uses the high level client.

There have been some corrupt producer message bugs related to using snappy
compression recently, but this does not seem to be the same as your problem.

Does MM stop on the exact same message each time (
https://cwiki.apache.org/confluence/display/KAFKA/System+Tools#SystemTools-ConsumerOffsetChecker
)? I would suggest triple checking that your configurations are the same
across all DC's (you mentioned that MM mirrors successfully to another DC
with no problem), as well as examine the problem message to see if you can
find anything different about it when compared to the others (See:
https://cwiki.apache.org/confluence/display/KAFKA/System+Tools#SystemTools-SimpleConsumerShell
). Your only other recourse is to iterate past the problem offset.

On Thu, Oct 1, 2015 at 1:22 AM, Jörg Wagner  wrote:

> Hey everyone,
>
> I've been having some issues with corrupted messages and mirrormaker as I
> wrote previously. Since there was no feedback, I want to ask a new question:
>
> Did you ever have corrupted messages in kafka? Did things break? How did
> you recover or work around that?
>
> Thanks
> Jörg
>

Retrieving last message offset in high level consumer

[eugene miretsky]

Hi,

We would like to log the offset of a Kafka message if we fail to process it
(so we can try to re-process it later). Is it possible to get the offset
using the high level consumer?

I took a quick look at the code, and:

   - It seems like the offset it private in the current Scala consumer
   (ConsumerIterator.scala)
   - The offset is exposed by the API of the new Java consumer
   (KafkaConsumer.java)


Would it make sense to expose the offset in the  Scala consumer as well?

Cheers,
Eugene

Shrinking ISR with no load on brokers or incoming messages

[Shaun Senecal]

Hi


I have noticed that when our brokers have no incoming connections (just 
connections to other brokers and to the ZK cluster) we get messages about 
shrinking the ISR for some partitions


[2015-10-02 00:58:31,239] INFO Partition [lia.stage.raw_events,9] on broker 1: 
Shrinking ISR for partition [lia.stage.raw_events,9] from 1,0,2 to 1 
(kafka.cluster.Partition)

...

[2015-10-02 00:58:31,335] INFO Partition [lia.stage.raw_events,9] on broker 1: 
Expanding ISR for partition [lia.stage.raw_events,9] from 1 to 1,0 
(kafka.cluster.Partition)

[2015-10-02 00:58:31,430] INFO Partition [lia.stage.raw_events,9] on broker 1: 
Expanding ISR for partition [lia.stage.raw_events,9] from 1,0 to 1,0,2 
(kafka.cluster.Partition)


It seems weird to me that the ISR would ever change when there is no load on 
the brokers and no incoming messages at all.  Does this indicate a problem with 
the cluster, or is this normal?




Thanks


Shaun

Re: Retrieving last message offset in high level consumer

[Karthikeyan Annamalai]

If i am not wrong, the auto commit might have happened so, when you start
the consumer it should work fine. Also keep it in mind that Kafka works on
at least one delivery model so we should expect redundant message while
restarting the consumer.
On Oct 2, 2015 4:06 AM, "eugene miretsky"  wrote:

> Hi,
>
> We would like to log the offset of a Kafka message if we fail to process it
> (so we can try to re-process it later). Is it possible to get the offset
> using the high level consumer?
>
> I took a quick look at the code, and:
>
>- It seems like the offset it private in the current Scala consumer
>(ConsumerIterator.scala)
>- The offset is exposed by the API of the new Java consumer
>(KafkaConsumer.java)
>
>
> Would it make sense to expose the offset in the  Scala consumer as well?
>
> Cheers,
> Eugene
>

RE: log clean up

[Karthikeyan Annamalai]

If its a daily rolling appender then next day when you get an another log
data will be rolled.
On Sep 26, 2015 3:20 AM, "Hema Bhatia"  wrote:

> Thanks Gwen!
>
> I made the changes and restarted kafka nodes. Looks like all log files are
> still present. Does it take some time to kick in the changes?
>
> Here is the sample of changes:
>
> log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
> log4j.appender.stateChangeAppender.DatePattern='.'-MM-dd-HH
> log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
> log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
> log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m
> (%c)%n
> log4j.appender.stateChangeAppender.MaxFileSize=100KB
> log4j.appender.stateChangeAppender.MaxBackupIndex=5
> .
> .
> log4j.appender.kafkaAppender.MaxFileSize=100KB
> log4j.appender.kafkaAppender.MaxBackupIndex=5
> .
> .
> log4j.appender.controllerAppender.MaxFileSize=100KB
> log4j.appender.controllerAppender.MaxBackupIndex=5
>
>
> -Original Message-
> From: Gwen Shapira [mailto:g...@confluent.io]
> Sent: Friday, September 25, 2015 1:13 PM
> To: users@kafka.apache.org
> Subject: Re: log clean up
>
> Absolutely.
>
> You can go into config/log4j.properties and configure the appenders to
> roll the logs.
>
> For example:
>
>
> log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
> log4j.appender.stateChangeAppender.DatePattern='.'-MM-dd-HH
> log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
> log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
> log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m
> (%c)%n
>
> log4j.appender.stateChangeAppender.MaxFileSize=100KB
> log4j.appender.stateChangeAppender.MaxBackupIndex=5
>
> Adding the last two lines will make sure you have 5 state change log files
> each 100KB.
>
> Gwen
>
> On Fri, Sep 25, 2015 at 11:04 AM, Hema Bhatia 
> wrote:
>
> > Is there a way to delete kafka server, controller and state-change logs.
> > They just keep growing over time and not purged.
> >
> > -Hema
> >
> >
> >
>