This post already has answers, but I am adding my view with a few pictures from Kafka Definitive Guide Before answering the questions, let’s look at an overview of producer components: 1. When a producer is producing a message – It will specify the topic it wants to send the message to, is that right? … Read more
In Kafka, the responsibility of what has been consumed is the responsibility of the consumer and this is also one of the main reasons why Kafka has such great horizontal scalability. Using the high level consumer API will automatically do this for you by committing consumed offsets in Zookeeper (or a more recent configuration option … Read more
To increase the number of replicas for a given topic you have to: 1. Specify the extra replicas in a custom reassignment json file For example, you could create increase-replication-factor.json and put this content in it: {“version”:1, “partitions”:[ {“topic”:”signals”,”partition”:0,”replicas”:[0,1,2]}, {“topic”:”signals”,”partition”:1,”replicas”:[0,1,2]}, {“topic”:”signals”,”partition”:2,”replicas”:[0,1,2]} ]} 2. Use the file with the –execute option of the kafka-reassign-partitions tool [or … Read more
I came across this question when trying to also consume from the __consumer_offsets topic. I managed to figure it out for different Kafka versions and thought I’d share what I’d found For Kafka 0.8.2.x Note: This uses Zookeeper connection #Create consumer config echo “exclude.internal.topics=false” > /tmp/consumer.config #Consume all offsets ./kafka-console-consumer.sh –consumer.config /tmp/consumer.config \ –formatter “kafka.server.OffsetManager\$OffsetsMessageFormatter” … Read more
You can increase Consumer poll() batch size by increasing max.partition.fetch.bytes, but still as per documentation it has limitation with fetch.max.bytes which also need to be increased with required batch size. And also from the documentation there is one other property message.max.bytes in Topic config and Broker config to restrict the batch size. so one way … Read more
Commits are handled by Streams internally and fully automatic, and thus there is usually no reason to commit manually. Note, that Streams handles this differently than consumer auto-commit — in fact, auto-commit is disabled for the internally used consumer and Streams manages commits “manually”. The reason is, that commits can only happen at certain points … Read more
Update Since Apache Kafka 2.1, offsets won’t be deleted as long as the consumer group is active, independent if the consumers commit offsets or not, ie, the offset.retention.minutes clocks only starts to tick when the group becomes empty (in older released, the clock started to tick directly when the commit happened). Cf. https://cwiki.apache.org/confluence/display/KAFKA/KIP-211%3A+Revise+Expiration+Semantics+of+Consumer+Group+Offsets Original Answer … Read more
Temporarily update the retention time on the topic to one second: kafka-topics.sh \ –zookeeper <zkhost>:2181 \ –alter \ –topic <topic name> \ –config retention.ms=1000 And in newer Kafka releases, you can also do it with kafka-configs –entity-type topics kafka-configs.sh \ –zookeeper <zkhost>:2181 \ –entity-type topics \ –alter \ –entity-name <topic name> \ –add-config retention.ms=1000 then … Read more
Before KIP-62, there is only session.timeout.ms (ie, Kafka 0.10.0 and earlier). max.poll.interval.ms is introduced via KIP-62 (part of Kafka 0.10.1). KIP-62, decouples heartbeats from calls to poll() via a background heartbeat thread, allowing for a longer processing time (ie, time between two consecutive poll()) than heartbeat interval. Assume processing a message takes 1 minute. If … Read more
This is because java is not recognizing root certificate authority (CA) SSL certificate of https://clojars.org/repo. Solution is to add that certificate to java cacerts file so that it got permanently accepted. Step 1 : Get root certificate of https://clojars.org Open https://clojars.org in a chrome browser. locate Lock symbol just besides your address bar and click … Read more