# Kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash\_logstash-indexer **URL:** \ **Category:** Logstash\ **Created:** [October 26, 2015, 6:40pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012 "2015-10-26T18:40:09Z")\ **Posts on this page:** 11\ **Page:** 1
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [October 26, 2015, 6:40pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/1 "2015-10-26T18:40:09Z")
I have an ELK Set-up in which a logstash is pushing data to Kafka and another logstash is pulling data from Kafka. Below are my Kafka Input Config:- ``` input { kafka { zk_connect => "kafka:2181" group_id => "logstash" topic_id => "logstash_logs" reset_beginning => false consumer_threads => 3 } } ``` I have gone through this [issue](https://github.com/joekiller/logstash-kafka/issues/16) & I have 3 partitions for my logstash topic. After starting the logstash I am seeing the below warnings and logstash is not pulling any data from Kafka:- ``` '[DEPRECATED] use `require 'concurrent'` instead of `require 'concurrent_ruby'` log4j, [2015-10-27T00:11:50.471] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1445884909915-d6a99924-2 for topic logstash_logs log4j, [2015-10-27T00:11:50.471] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1445884909915-d6a99924-0 for topic logstash_logs log4j, [2015-10-27T00:11:50.471] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1445884909915-d6a99924-1 for topic logstash_logs Logstash startup completed ``` Can someone let me know what else can go wrong? Environment - - Logstash - 1.5.3 - Kafka - 0.8.2.1 ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [October 27, 2015, 12:30pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/2 "2015-10-27T12:30:55Z")
The Kafka consumer starts at the largest offset by default from when the consumer group is created. If you want data that is older you have to remove the consumer group and then set the offset default to smallest for the initial pull. You can trigger it by starting one logstash-kafka-input instance and use the [https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-reset\_beginning](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-reset_beginning) setting. ---
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [October 27, 2015, 12:38pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/3 "2015-10-27T12:38:51Z")
Thanks @Joe_Lawson. You meant to say I should restart the logstash instance by setting reset\_beginning to true? ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [October 27, 2015, 4:47pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/4 "2015-10-27T16:47:16Z")
Yeah update you logstash config with reset\_beginning true and it'll reset the consumer group offsets in zookeeper so after if does that, stop the logstash service and take it out. Otherwise it would reset it every time which you don't want. ---
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [October 28, 2015, 9:11am UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/5 "2015-10-28T09:11:30Z")
@Joe_Lawson - As you suggested I set the `reset_beginning => true` started the logstash and when I saw the message `Logstash startup Completed` I stopped the logstash and then restarted it again with `reset_beginning => false`. It worked but after some time I observed that logstsash is not again pulling message from Kafka on restarting I saw the same WARNING message. ``` WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1445884909915-d6a99924-2 for topic logstash_logs ``` Do you have any idea why is logstash suddenly stopping to pull message from Kafka? ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [October 28, 2015, 9:34pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/6 "2015-10-28T21:34:32Z")
The error you are reporting implies that there are less partitions than threads. Using the kafka tools, run bin/kafka-topics.sh --describe --topic logstash\_logs --zookeeper kafka:2181 That should tell you how many partitions are on the topic. Also set consumer\_threads to 1. You don't gain anything setting it larger. ---
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [October 29, 2015, 6:11am UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/7 "2015-10-29T06:11:58Z")
@Joe_Lawson - I am having 3 partitions that I verified. Below is the output:- ``` root@kafka:/opt/kafka_2.10-0.8.2.1# bin/kafka-topics.sh --describe --topic logstash_logs --zookeeper localhost:2181 Topic:logstash_logs PartitionCount:3 ReplicationFactor:1 Configs: Topic: logstash_logs Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: logstash_logs Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Topic: logstash_logs Partition: 2 Leader: 0 Replicas: 0 Isr: 0 ``` As per the [documentation](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-consumer_threads) no. of threads should be equal to the no. of partitions. When should we set this to more than 1 then? ---
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [November 1, 2015, 2:06pm UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/8 "2015-11-01T14:06:41Z")
I created a new topic `logstash_logs2` but seeing the same problem ☹ . I started logstash with `--verbose`. Below are the logs:- ``` log4j, [2015-11-01T19:32:37.452] INFO: org.apache.zookeeper.ClientCnxn: Session establishment complete on server kafka/172.16.84.188:2181, sessionid = 0x150a82d75a55c66, negotiated timeout = 6000 log4j, [2015-11-01T19:32:37.463] INFO: org.I0Itec.zkclient.ZkClient: zookeeper state changed (SyncConnected) log4j, [2015-11-01T19:32:37.473] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], starting auto committer every 1000 ms log4j, [2015-11-01T19:32:37.511] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], begin registering consumer logstash_logstash-indexer-1446386557328-9606786c in ZK log4j, [2015-11-01T19:32:37.584] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], end registering consumer logstash_logstash-indexer-1446386557328-9606786c in ZK log4j, [2015-11-01T19:32:37.598] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], starting watcher executor thread for consumer logstash_logstash-indexer-1446386557328-9606786c log4j, [2015-11-01T19:32:37.663] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], begin rebalancing consumer logstash_logstash-indexer-1446386557328-9606786c try #0 log4j, [2015-11-01T19:32:37.875] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1446386557468] Stopping leader finder thread log4j, [2015-11-01T19:32:37.875] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1446386557468] Stopping all fetchers log4j, [2015-11-01T19:32:37.878] INFO: kafka.consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1446386557468] All connections stopped log4j, [2015-11-01T19:32:37.878] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], Cleared all relevant queues for this fetcher log4j, [2015-11-01T19:32:37.880] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], Cleared the data chunks in all the consumer message iterators log4j, [2015-11-01T19:32:37.881] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], Committing all offsets after clearing the fetcher queues log4j, [2015-11-01T19:32:37.883] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], Releasing partition ownership log4j, [2015-11-01T19:32:37.958] INFO: kafka.consumer.RangeAssignor: Consumer logstash_logstash-indexer-1446386557328-9606786c rebalancing the following partitions: ArrayBuffer(0, 1, 2) for topic logstash_logs2 with consumers: List(logstash_logstash-indexer-1446358311049-e6b11928-0, logstash_logstash-indexer-1446358311049-e6b11928-1, logstash_logstash-indexer-1446358311049-e6b11928-2, logstash_logstash-indexer-1446358341201-b911cc94-0, logstash_logstash-indexer-1446358341201-b911cc94-1, logstash_logstash-indexer-1446358341201-b911cc94-2, logstash_logstash-indexer-1446364131441-49f4f931-0, logstash_logstash-indexer-1446364131441-49f4f931-1, logstash_logstash-indexer-1446364131441-49f4f931-2, logstash_logstash-indexer-1446386557328-9606786c-0) log4j, [2015-11-01T19:32:37.960] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1446386557328-9606786c-0 for topic logstash_logs2 log4j, [2015-11-01T19:32:37.998] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], Consumer logstash_logstash-indexer-1446386557328-9606786c selected partitions : log4j, [2015-11-01T19:32:38.001] INFO: kafka.consumer.ZookeeperConsumerConnector: [logstash_logstash-indexer-1446386557328-9606786c], end rebalancing consumer logstash_logstash-indexer-1446386557328-9606786c try #0 log4j, [2015-11-01T19:32:38.004] INFO: kafka.consumer.ConsumerFetcherManager$LeaderFinderThread: [logstash_logstash-indexer-1446386557328-9606786c-leader-finder-thread], Starting ``` I tried with `consumer_thread => 1` but it does not help much. ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [November 4, 2015, 3:36am UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/9 "2015-11-04T03:36:35Z")
How many logstash processes do you have connecting to those brokers that are using the same consumer group I'd? From the following snippit it looks like you have four different logstash consumers hitting that topic all in the same group. Three consumers have three threads and one has a single thread. That makes 10 consumer threads for only three partitions so 7 are going to throw errors. For each consumer group you can only have as many consumer threads as there are partitions. ```auto log4j, [2015-11-01T19:32:37.958] INFO: kafka.consumer.RangeAssignor: Consumer logstash_logstash-indexer-1446386557328-9606786c rebalancing the following partitions: ArrayBuffer(0, 1, 2) for topic logstash_logs2 with consumers: List(logstash_logstash-indexer-1446358311049-e6b11928-0, logstash_logstash-indexer-1446358311049-e6b11928-1, logstash_logstash-indexer-1446358311049-e6b11928-2, logstash_logstash-indexer-1446358341201-b911cc94-0, logstash_logstash-indexer-1446358341201-b911cc94-1, logstash_logstash-indexer-1446358341201-b911cc94-2, logstash_logstash-indexer-1446364131441-49f4f931-0, logstash_logstash-indexer-1446364131441-49f4f931-1, logstash_logstash-indexer-1446364131441-49f4f931-2, logstash_logstash-indexer-1446386557328-9606786c-0) log4j, [2015-11-01T19:32:37.960] WARN: kafka.consumer.RangeAssignor: No broker partitions consumed by consumer thread logstash_logstash-indexer-1446386557328-9606786c-0 for topic logstash_logs2 ``` ---
**Author:** ![Deb](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/deb/32/46162_2.png) [@Deb](https://discuss.elastic.co/u/Deb)\ **Post date:** [November 5, 2015, 10:31am UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/10 "2015-11-05T10:31:13Z")
There were some other logstash instances running which I was not aware of. Thanks again @Joe_Lawson for pointing this out. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 5:23am UTC](https://discuss.elastic.co/t/kafka-consumer-rangeassignor-no-broker-partitions-consumed-by-consumer-thread-logstash-logstash-indexer/33012/11 "2017-07-06T05:23:56Z")