Consumers groups is a Kafka abstraction that enables supporting both point-to-point and publish/subscribe messaging. A consumer can join a consumer group (let us say `group_1`) by setting its `group.id` to `group_1`. Consumer groups is also a way of supporting parallel consumption of the data i.e. different consumers of the same consumer group consume data in parallel from different partitions.

In addition to group.id, each consumer also identifies itself to the Kafka broker using `consumer.id`. This is used by Kafka to identify the currently ACTIVE consumers of a particular consumer group.

Read [this documentation][1] for more details.


  [1]: http://kafka.apache.org/documentation.html

How to delete a never-use-again file and release the storage back to the github lfs quota?

Does deleting the reference point of the file in git history work in this situation?

How to delete a file tracked by git-lfs and release the storage quota?

I am looking for an example in handling Bubbling and Capturing in React.js. I found [one with JavaScript][1], but I am having trouble finding the equivalent for React.js.

How would I have to create an example for Bubbling and Capturing in React.js?


  [1]: https://stackoverflow.com/questions/4616694/what-is-event-bubbling-and-capturing

Example for Bubbling and Capturing in React.js

I noticed that the Consumer configuration has two IDs. One is `group.id` (mandatory) and second one is `consumer.id` (not Mandatory). 

What is the difference between these 2 IDs?

difference between groupid and consumerid in Kafka consumer

<p>I noticed that the Consumer configuration has two IDs. One is <code>group.id</code> (mandatory) and second one is <code>consumer.id</code> (not Mandatory).</p>
<p>What is the difference between these 2 IDs?</p>


How can I get number of partitions for any kafka topic from the code. I have researched many links but none seem to work.

Mentioning a few:

http://grokbase.com/t/kafka/users/148132gdzk/find-topic-partition-count-through-simpleclient-api

http://grokbase.com/t/kafka/users/151cv3htga/get-replication-and-partition-count-of-a-topic

http://qnalist.com/questions/5809219/get-replication-and-partition-count-of-a-topic

which look like similar discussions.

Also there are similar links on SO which do not have a working solution to this.

kafka get partition count for a topic

We are planning to write a Kafka consumer(java) which reads Kafka queue to perform an action which is in the message.

As the consumers run independently, will the message is processed by only one consumer at a time? Else all the consumers process the same message as they have their own offset in the partition.

Please help me understand.

Can multiple Kafka consumers read same message from the partition

I am trying to use Kafka.  
All configurations are done properly but when I try to produce message from console I keep getting the following error

    WARN Error while fetching metadata with correlation id 39 : 
         {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)


Kafka version: `2.11-0.9.0.0`

Leader Not Available Kafka in Console Producer

I am trying to use `wurstmeister\kafka-docker` image with  `docker-compose`, but I am having real problems with connecting everything.

All the posts or questions that I check, seems not to have any problems, but I am frankly lost. (And there are at least two questions in SO that try to address the problem)

I believe that the problem is my poor understanding of the networking of `docker`. So the problem:

I can consume and produce from the same container of kafka, but, when I try to create another container (or use my laptop with a python client) I got several errors related to the `advertised.host.name` parameter (in the image this parameter is `KAFKA_ADVERTISED_HOST_NAME`) 

I already try setting this variable in lot of ways, but it simply don&#39;t work.

So I am looking for a authorative answer (i.e. how to set automatically those parameters and what is it meaning)  how to set the `docker-compose.yml`

This is mine:

    zookeeper:
      image: wurstmeister/zookeeper
      ports:
        - &quot;2181:2181&quot;
    
    kafka:
      image: wurstmeister/kafka
     # hostname: kafka
      ports:
        - &quot;9092&quot;
      links:
        - zookeeper:zk
      environment:
        KAFKA_ADVERTISED_HOST_NAME: &quot;kafka&quot;
        KAFKA_ADVERTISED_PORT: &quot;9092&quot;
        KAFKA_ZOOKEEPER_CONNECT: &quot;zk:2181&quot;


**UPDATE**

Following the advise of @dnephin, I modified the `start-kafka.sh` in the following lines:

    ...
    if [[ -z &quot;$KAFKA_ADVERTISED_PORT&quot; ]]; then
        export KAFKA_ADVERTISED_PORT=$(hostname -i)
    fi
    ...

and  remove `KAFKA_ADVERTISED_HOST_NAME: &quot;kafka&quot;` from the `docker-compose.yml` 


I started the containers in the canonical way:

    docker-compose up -d

Both of the containers are running:

    $ docker-compose ps
               Name                          Command               State                     Ports                    
    -----------------------------------------------------------------------------------------------------------------
    infraestructura_kafka_1       start-kafka.sh                   Up      0.0.0.0:32768-&gt;9092/tcp                    
    infraestructura_zookeeper_1   /opt/zookeeper/bin/zkServe ...   Up      0.0.0.0:2181-&gt;2181/tcp, 2888/tcp, 3888/tcp 

Afterwards I did:

    docker-compose logs

And everything run smoothly.

For checking the ip addresses:


    $ KAFKA_IP=$(docker inspect --format &#39;{{ .NetworkSettings.IPAddress }}&#39; infraestructura_kafka_1)                                                                                                            
    $ echo $KAFKA_IP
    172.17.0.4
    
    and
    
    $ ZK_IP=$(docker inspect --format &#39;{{ .NetworkSettings.IPAddress }}&#39; infraestructura_zookeeper_1)                                                                                                           
    $ echo $ZK_IP 
    172.17.0.3


Then I execute in two differents consoles:

A producer:

    $ docker run --rm --interactive wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092  

A consumer:

    $ docker run --rm --interactive  wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --topic grillo --from-beginning --zookeeper 172.17.0.3:2181 

Almost immediately, warnings start flying all over the screen:

    [2016-03-11 00:39:17,010] WARN Fetching topic metadata with correlation id 0 for topics [Set(grillo)] from broker [BrokerEndPoint(1001,ba53d4fd7595,9092)] failed (kafka.client.ClientUtils$)
    java.nio.channels.ClosedChannelException
            at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
            at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
            at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
            at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
            at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
            at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
            at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
            at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
    [2016-03-11 00:39:17,013] WARN [console-consumer-79688_9dd5f575d557-1457656747003-f1ed369d-leader-finder-thread], Failed to find leader for Set([grillo,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFin
    derThread)
    kafka.common.KafkaException: fetching topic metadata for topics [Set(grillo)] from broker [ArrayBuffer(BrokerEndPoint(1001,ba53d4fd7595,9092))] failed
            at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
            at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
            at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
            at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
    Caused by: java.nio.channels.ClosedChannelException
            at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
            at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
            at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
            at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
            at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
            ... 3 more

and so on


In the console of the producer, I wrote some sentences:

    $ docker run --rm --interactive klustera/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092                                                           
    Hola
    &#191;C&#243;mo est&#225;ń?
    &#191;Todo bien?

And a few moments later, I got this response:

    [2016-03-11 00:39:28,955] ERROR Error when sending message to topic grillo with key: null, value: 4 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    [2016-03-11 00:40:28,956] ERROR Error when sending message to topic grillo with key: null, value: 16 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
    [2016-03-11 00:41:28,956] ERROR Error when sending message to topic grillo with key: null, value: 12 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)

And in the `docker-compose logs`

    ...
    zookeeper_1 | 2016-03-11 00:39:07,072 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
    xid:0x2 zxid:0x47 txntype:-1 reqpath:n/a Error Path:/consumers Error:KeeperErrorCode = NodeExists for /consumers
    zookeeper_1 | 2016-03-11 00:39:07,243 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
    xid:0x19 zxid:0x4b txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners/grillo Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners/grillo
    zookeeper_1 | 2016-03-11 00:39:07,247 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create $xid:0x1a zxid:0x4c txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners
    ...


**UPDATE 2**

I made it work, at least, in `docker-machine`:

First, I defined a variable with the name of the `docker-machine`:

    DOCKER_VM=kafka_test

Then, I modify the `docker-compose.yml` as follows:

    KAFKA_ADVERTISED_HOST_NAME: &quot;${DOCKER_MACHINE_IP}&quot;

Lastly, in the environment of the `docker-machine`, I execute:

    DOCKER_MACHINE_IP=$(docker-machine ip $DOCKER_VM) docker-compose up -d


But in the laptop (I mean, without using a virtual machine, it doesn&#39;t work)



Kafka in Docker not working

Kafka splits incoming messages up into partitions, according to the partition assigned by the producer. Messages from partitions then get consumed by consumers in different consumer groups. 

This architecture makes me wary of using Kafka as a work/task queue, because I have to specify the partition at time of production, which indirectly limits which consumers can work on it because a partition is sent to only one consumer in a consumer group. I would rather not specify the partition ahead of time, so that whichever consumer is available to take that task can do so. Is there a way to structure partitions/producers in a Kafka architecture where tasks can be pulled by the next available consumer, without having to split up work ahead of time by choosing a partition when the work is produced?

Using only one partition for this topic would put all the tasks in the same queue, but then the number of consumers is limited to 1 per consumer group, so each consumer would have to be in a different group. Then all of the task get distributed to each consumer group, though, which is not the kind of work queue I&#39;m looking for.

Is Apache Kafka appropriate for use as a task queue?

Is Apache Kafka appropriate for use as an unordered task queue?

I am starting to learn Kafka, during my readings, some questions came to my mind:

 1. When a producer is producing a message - it will specify the **topic** it wants to send the message to, is that right? Does it care about partitions?
 2. When a subscriber is running - does it specify its group id so that it can be part of a cluster of consumers of the same topic or several topics that this group of consumers is interested in?
 3. Does each consumer group have a corresponding partition on the broker or does each consumer have one?
 
 4. Are the partitions created by the broker, and therefore not a concern for the consumers?
 
 5. Since this is a queue with an offset for each partition, is it the responsibility of the consumer to specify which messages it wants to read? Does it need to save its state?
 
 6. What happens when a message is deleted from the queue? - For example, the retention was for 3 hours, then the time passes, how is the offset being handled on both sides?


Understanding Kafka Topics and Partitions

I am unclear why we need both `session.timeout.ms` and `max.poll.interval.ms` and when would we use one or the other or both? It seems like both settings indicate the upper bound on the time the coordinator will wait to get the heartbeat from a consumer before assuming it&#39;s dead.

Also how does it behave for versions 0.10.1.0+ based on [KIP-62][1]?


  [1]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread

Difference between session.timeout.ms and max.poll.interval.ms for Kafka &gt;= 0.10.1

I recently started learning Kafka and end up with these questions. 

 1. What is the difference between Consumer and Stream? For me, if any tool/application consume messages from Kafka is a consumer in the Kafka world. 

 2. How Stream is different as this also consumes from or produce messages to Kafka? and why is it needed as we can write our own consumer 
application using Consumer API and process them as needed or send them to Spark from the consumer application?

I did Google on this, but did not get any good answers for this. Sorry if this question is too trivial.


Kafka: Consumer API vs Streams API

Nowadays in microservice world, i’m seeing alot of design in my workplace that uses kafka messaging when you can achieve similar results using rest api calls between microservices. Technically you can stop using rest api calls altogether and instead use kafka messaging. I really want to know the best practice, pros and cons, when to use api calls between microsevices, when to use kafka messaging. 


Lets put a real life example:

I have an inventory service and a vendor service. Everyday vendor service calls the vendor API to get new items and these need to be moved into inventory service. The number of items can be up to 10,000 objects.

For this use case, is it better to :

1. After getting new data from vendor API, call REST API of inventory service to store the new items.

2. After getting new data from vendor API, send them as message to a kafka topic, to be consumed by inventory service 

Which way would you choose and what is the consideration

Content Type	Original Author	Original Content on Stackoverflow
Question	Gnana	View Question on Stackoverflow
Solution 1 - Apache Kafka	Aravind Yarram	View Answer on Stackoverflow

difference between groupid and consumerid in Kafka consumer

Apache Kafka Problem Overview

Apache Kafka Solutions

Solution 1 - Apache Kafka

Example for Bubbling and Capturing in React.js

How to delete a file tracked by git-lfs and release the storage quota?

Attributions