Re: Kafka won't replicate from a specific broker

[Jan Omar]

Unfortunately I think you hit this bug:

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


The only option I know of is to reboot the affected broker. And upgrade to 
0.10.1.1 as quickly as possible. We haven't seen this issue on 0.10.1.1.RC0.

Regards

Jan


> On 22 Dec 2016, at 18:16, Ismael Juma  wrote:
> 
> Hi Valentin,
> 
> Is inter.broker.protocol.version set correctly in brokers 1 and 2? It
> should be 0.10.0 so that they can talk to the older broker without issue.
> 
> Ismael
> 
> On Thu, Dec 22, 2016 at 4:42 PM, Valentin Golev 
> wrote:
> 
>> Hello,
>> 
>> I have a three broker Kafka setup (the ids are 1, 2 (kafka 0.10.1.0) and
>> 1001 (kafka 0.10.0.0)). After a failure of two of them a lot of the
>> partitions have the third one (1001) as their leader. It's like this:
>> 
>> Topic: userevents0.open Partition: 5Leader: 1   Replicas:
>> 1,2,1001  Isr: 1,1001,2
>>Topic: userevents0.open Partition: 6Leader: 2   Replicas:
>> 2,1,1001  Isr: 1,2,1001
>>Topic: userevents0.open Partition: 7Leader: 1001Replicas:
>> 1001,2,1  Isr: 1001
>>Topic: userevents0.open Partition: 8Leader: 1   Replicas:
>> 1,1001,2  Isr: 1,1001,2
>>Topic: userevents0.open Partition: 9Leader: 1001Replicas:
>> 2,1001,1  Isr: 1001
>>Topic: userevents0.open Partition: 10   Leader: 1001Replicas:
>> 1001,1,2  Isr: 1001
>> 
>> As you can see, only the partitions with Leaders 1 or 2 have successfully
>> replicated. Brokers 1 and 2, however, are unable to fetch data from the
>> 1001.
>> 
>> All of the partitions are available to the consumers and producers. So
>> everything is fine except replication. 1001 is available from the other
>> servers.
>> 
>> I can't restart the broker 1001 because it seems that it will cause data
>> loss (as you can see, it's the only ISR on many partitions). Restarting the
>> other brokers didn't help at all. Neither did just plain waiting (it's the
>> third day of this going on). So what do I do?
>> 
>> The logs of the broker 2 (the one which tries to fetch data) are full of
>> this:
>> 
>> [2016-12-22 16:38:52,199] WARN [ReplicaFetcherThread-0-1001], Error in
>> fetch kafka.server.ReplicaFetcherThread$FetchRequest@117a49bf
>> (kafka.server.ReplicaFetcherThread)
>> java.io.IOException: Connection to 1001 was disconnected before the
>> response was read
>>at
>> kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$
>> extension$1$$anonfun$apply$1.apply(NetworkClientBlockingOps.scala:115)
>>at
>> kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$
>> extension$1$$anonfun$apply$1.apply(NetworkClientBlockingOps.scala:112)
>>at scala.Option.foreach(Option.scala:257)
>>at
>> kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$
>> extension$1.apply(NetworkClientBlockingOps.scala:112)
>>at
>> kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$
>> extension$1.apply(NetworkClientBlockingOps.scala:108)
>>at
>> kafka.utils.NetworkClientBlockingOps$.recursivePoll$1(
>> NetworkClientBlockingOps.scala:137)
>>at
>> kafka.utils.NetworkClientBlockingOps$.kafka$utils$
>> NetworkClientBlockingOps$$pollContinuously$extension(
>> NetworkClientBlockingOps.scala:143)
>>at
>> kafka.utils.NetworkClientBlockingOps$.blockingSendAndReceive$extension(
>> NetworkClientBlockingOps.scala:108)
>>at
>> kafka.server.ReplicaFetcherThread.sendRequest(ReplicaFetcherThread.scala:
>> 253)
>>at
>> kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:238)
>>at
>> kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:42)
>>at
>> kafka.server.AbstractFetcherThread.processFetchRequest(
>> AbstractFetcherThread.scala:118)
>>at
>> kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:103)
>>at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
>> 
>> The logs of the broker 1001 are full of this:
>> 
>> [2016-12-22 16:38:54,226] ERROR Processor got uncaught exception.
>> (kafka.network.Processor)
>> java.nio.BufferUnderflowException
>>at java.nio.Buffer.nextGetIndex(Buffer.java:506)
>>at java.nio.HeapByteBuffer.getInt(HeapByteBuffer.java:361)
>>at
>> kafka.api.FetchRequest$$anonfun$1$$anonfun$apply$1.
>> apply(FetchRequest.scala:55)
>>at
>> kafka.api.FetchRequest$$anonfun$1$$anonfun$apply$1.
>> apply(FetchRequest.scala:52)
>>at
>> scala.collection.TraversableLike$$anonfun$map$
>> 1.apply(TraversableLike.scala:234)
>>at
>> scala.collection.TraversableLike$$anonfun$map$
>> 1.apply(TraversableLike.scala:234)
>>at scala.collection.immutable.Range.foreach(Range.scala:160)
>>at
>> scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
>>at 

Consumer failover issue when coordinator dies (e.g. broker restart)

[Jan Omar]

Hey guys,

We're running Kafka 0.9.0.1 with Java 7 on FreeBSD. We are experiencing 
unrecoverable issues in our consumers, e.g. when restarting brokers.

The consumers start reporting that the coordinator died (which in general is 
correct, because the coordinator was restarted). However, the consumer should 
failover to another coordinator, unfortunately that never happens. Instead it 
runs into an inifite loop, that looks like this:

[exa3-fetcher] INFO 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Marking the 
coordinator 2147483641 dead.
[exa3-fetcher] INFO 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Attempt to 
join group bit-exa3 failed due to obsolete coordinator information, retrying.

It remains like this until we shutdown all brokers except for 1. As soon as we 
have only 1 broker remaining life continues as expected.

[exa3-fetcher] INFO 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Attempt to 
heart beat failed since the group is rebalancing, try to re-join group.

Any idea what's causing this or if this a known bug or something?

Any help would be highly appreciated as this is a serious issue for us.

Thanks!

Regards

Jan

Re: To find the Lag of consumer offset using kafka client library

[Jan Omar]

Hi Gourab,

Check this out:

https://github.com/linkedin/Burrow 

Regards

Jan

> On 29 Sep 2016, at 15:47, Gourab Chowdhury  wrote:
> 
> I can get the *Lag* of offsets with the following command:-
> 
> bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand --zookeeper
> localhost:2182 --describe --group DemoConsumer
> 
> I am trying to find code that uses kafka library to find the *Lag* of
> offsets in consumer?
> 
> Also is there any other documentation other than https://github.com/apache/
> kafka/tree/trunk/docs? I can't find much documentation of kafka.
> 
> Thanks,
> Gourab Chowdhury,
> Software Engg. JunctionTV Inc.

Re: Offset Tools

[Jan Omar]

Hi Joerg,

We recently built a tool for fetching current offsets (per partition) for a 
given consumer group. Also for setting the group to a specific offset.
It even allows to reset a consumer group to a given timestamp by running bisect 
(binary search) on the messages. 

Unfortunately we're using a proprietary message format, that's why we don't 
have any plans (or capacity) to open source it at the moment.

However builiding that tool was straight forward, it shouldn't take you more 
than a day or two to build something similar. Ping me if you need some help.

Regards

Jan 

> On 11 Jul 2016, at 13:00, Jörg Wagner  wrote:
> 
> Hello!
> 
> We recently switched to Kafka 0.9.0.1 and currently I don't seem to be able 
> to figure out how to read the consumer offsets via cli. We are using the 
> 0.9.0.1 new consumer and are storing the offsets in kafka.
> 
> Status:
> kafka-consumer-offset-checker.sh is old and deprecated, points to 
> kafka-consumer-groups.sh
> kafka-consumer-groups.sh in old consumer mode shows lots of data, but unknown 
> for offsets and lag
> kafka-consumer-groups.sh in new consumer mode shows only one single consumer 
> group, not the one(s) expected. The one available though shows all the 
> correct data I would like to see for a group not shown.
> 
> Short: since 0.9.0.1 I can't see the consumer offsets anymore.
> 
> Anyone else experiencing this?
> 
> Cheers
> Jörg

Re: [DISCUSS] Java 8 as a minimum requirement

[jan . omar]

Hi Ismael,

Unfortunately Java 8 doesn't play nice with FreeBSD. We have seen a lot of JVM 
crashes running our 0.9 brokers on Java 8... Java 7 on the other hand is 
totally stable. 

Until these issues have been addressed, this would cause some serious issues 
for us.

Regards 

Jan

Re: Three consumers on a single partition

[Jan Omar]

Hi Rami, 

Each consumer will receive every single message if they belong to different 
consumer groups. Messages will only be distributed between consumers of the 
same consumer group.

So make sure they are in the same consumer group, beware in your case this 
means 2 of the 3 consumers will be starving. To solve this issue you need to 
increase your partition count.

Regards

Jan

> On 14 Jun 2016, at 13:07, Joris Peeters  wrote:
> 
> I suppose the consumers would also need to all belong to the same consumer 
> group for your expectation to hold. If the three consumers belong to 
> different consumer groups, I'd expect each of them to receive all the 
> messages, regardless of the number of partitions.
> So perhaps they are on different consumer groups? What client are you using?
> 
> -Original Message-
> From: thinking [mailto:letianyi...@qq.com]
> Sent: 14 June 2016 12:04
> To: users 
> Subject: Re: Three consumers on a single partition
> 
> hi,
>   1. you should check, if your topic partition really have one.
>   2. does your consumer get same message or different message.
>e.x all message is 1,2,3,4,5,6,7.  consumer1 get 1,3,7 consumer2 get 2,6, 
> consumer3 get 4,5?
> 
> 
> 
> 
> -- Original --
> From:  "Al-Isawi Rami";;
> Date:  Tue, Jun 14, 2016 06:58 PM
> To:  "users@kafka.apache.org";
> 
> Subject:  Three consumers on a single partition
> 
> 
> 
> Hi,
> 
> I have a cluster of 3 brokers and 1 topic which has 1 partition and 
> replication factor of 3. There are also 3 consumers consuming from that topic.
> 
> Now all the docs I have seen say that if number of consumers is bigger than 
> the number of partition ( like in my case 3 consumers 1 partition), then only 
> one consumer will get messages and the other two will not get any, 
> nonetheless, they are all connected.
> 
> However, in my case, all the three consumers are getting messages. Any ideas 
> why this is happening?
> 
> Regards,
> -Rami
> Disclaimer: This message and any attachments thereto are intended solely for 
> the addressed recipient(s) and may contain confidential information. If you 
> are not the intended recipient, please notify the sender by reply e-mail and 
> delete the e-mail (including any attachments thereto) without producing, 
> distributing or retaining any copies thereof. Any review, 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. Thank 
> you.
> 
> 
> Winton Capital Management Limited (“Winton”) is a limited company registered 
> in England and Wales with its registered offices at 16 Old Bailey, London, 
> EC4M 7EG (Registered Company No. 3311531). Winton is authorised and regulated 
> by the Financial Conduct Authority in the United Kingdom, registered as an 
> investment adviser with the US Securities and Exchange Commission, registered 
> with the US Commodity Futures Trading Commission and a member of the National 
> Futures Association in the United States.
> 
> This communication, including any attachments, is confidential and may be 
> privileged. This email is for use by the intended recipient only. If you 
> receive it in error, please notify the sender and delete it. You should not 
> copy or disclose all or any part of this email.
> 
> This email does not constitute an offer or solicitation and nothing contained 
> in this email constitutes, and should not be construed as, investment advice. 
> Prospective investors should request offering materials and consult their own 
> advisers with respect to investment decisions and inform themselves as to 
> applicable legal requirements, exchange control regulations and taxes in the 
> countries of their citizenship, residence or domicile. Past performance is 
> not indicative of future results.
> 
> Winton takes reasonable steps to ensure the accuracy and integrity of its 
> communications, including emails. However Winton accepts no liability for any 
> materials transmitted. Emails are not secure and cannot be guaranteed to be 
> error free. Winton handles personal information in accordance with its 
> privacy 
> notice.

Re: Can we delete topic in kafka

[Jan Omar]

You have to allow topic deletion in server.properties first.

delete.topic.enable = true

Regards

Jan

> On 11 May 2016, at 09:48, Snehalata Nagaje 
>  wrote:
> 
> 
> 
> Hi , 
> 
> Can we delete certain topic in kafka? 
> 
> I have deleted using command 
> 
> ./kafka-topics.sh --delete --topic topic_billing --zookeeper localhost:2181 
> 
> It says topic marked as deletion, but it does not actually delete topic. 
> 
> Thanks, 
> Snehalata 

Re: How to deserialize the object without avro schema?

[jan . omar]

Hi, avro Schemas imply a pretty big overhead, if you would include them in 
every message. It's good practice to include a schema id with the message... 
Then you need a schema repository to lookup the matching schema based on the id.

Have a look at confluent.io. They offer a schema repo among other Kafka related 
tools.

Regards 

Jan



Sent from my iPhone
> On 19 Apr 2016, at 08:02, Ratha v  wrote:
> 
> Hi all;
> 
> I try to publish/consume my java objects to kafka. I use Avro schema.
> 
> My basic program works fine. In my program i use my schema in the producer
> (for encoding) and consumer (decoding).
> 
> If i publish different objects to different topics( eg: 100 topics)at the
> receiver, i do not know, what type of message i received. I would like to
> get the avro schema from the received byte and would like to use that for
> decoding. Is that right? If so, how can i retrieve from the received object?
> Or is there any better approach?
> 
> Thanks.
> -- 
> -Ratha
> http://vvratha.blogspot.com/

Re: Consumers doesn't always poll first messages

[Jan Omar]

Hi Robin,

Why would you expect it to start from the first message?

You're comitting the read offsets automatically every second. The offset is 
persisted, next time you consume again, it will start at the persisted offset 
again.

 consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
 consumerProperties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

Regards

Jan

> On 2 Mar 2016, at 15:14, Péricé Robin  wrote:
> 
> Hello everybody,
> 
> I'm testing the new 0.9.0.1 API and I try to make a basic example working.
> 
> *Java code* :
> 
> *Consumer *: http://pastebin.com/YtvW0sz5
> *Producer *: http://pastebin.com/anQay9YE
> *Test* : http://pastebin.com/nniYLsHL
> 
> 
> *Kafka configuration* :
> 
> *Zookeeper propertie*s : http://pastebin.com/KC5yZdNx
> *Kafka properties* : http://pastebin.com/Psy4uAYL
> 
> But when I try to run my test and restart Kafka to see what happen. The
> Consumer doesn't always consume first messages. Sometimes it consume
> messages at offset 0 or 574 or 1292 ... The behavior of the test seems to
> be very random.
> 
> Anybody have an idea on that issue ?
> 
> Best Regards,
> 
> Robin

Kafka Rest Proxy

[Jan Omar]

Hey guys, 

Is someone using the kafka rest proxy from confluent? 

We have an issue, that all messages for a certain topic end up in the same 
partition. Has anyone faced this issue before? We're not using a custom 
  partitioner class, so it's using the default partitioner. We're sending 
messages without a specific partition and without a key, like this: 
  

curl -X POST -H "Content-Type: application/vnd.kafka.json.v1+json" --data 
'{"records":[{"value":{"foo":"bar"}}]}' "http://x.x.x.x:8082/topics/testme; 


and yet for some reason every message ends up in partition 11... 

It's a test topic with 30 partitions on 3 brokers and our rest (producer) 
config is very simple:

id=1
zookeeper.connect=zookeeper-four.acc:2181...etc

Any help would be appreciated.

Thanks!