[ https://issues.apache.org/jira/browse/STORM-2342?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Felipe Cavalcanti updated STORM-2342: ------------------------------------- Description: I've created a topology that will read from kafka using storm-kafka-client but when it reaches the last message on kafka log it stops consuming and get stuck, new messages are never consumed, here are the kafka logs: [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Preparing to restabilize group kafka-spout with old generation 29 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Stabilized group kafka-spout generation 30 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:15:26,868] INFO [GroupCoordinator 1002]: Assignment received from leader for group kafka-spout for generation 30 (kafka.coordinator.GroupCoordinator) ========= here storm starts consuming messages, then, when it hits the last message, I can see this log in kafka == > [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Preparing to restabilize group kafka-spout with old generation 30 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Group kafka-spout with generation 31 is now empty (kafka.coordinator.GroupCoordinator) ===== and then storm consumer group is stuck, no new messages are read from kafka. my topology/ spout are configured that way: Topology: c.put(SConfig.TOPOLOGY_MAX_SPOUT_PENDING, 1000) c.put(SConfig.NIMBUS_SEEDS, "my nimbus seeds") c.put(SConfig.NIMBUS_THRIFT_PORT, 6627) c.put(SConfig.TOPOLOGY_WORKERS, 2) c.put(SConfig.TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS, 100) Spout: props.put(KafkaSpoutConfig.Consumer.ENABLE_AUTO_COMMIT, true) props.put(KafkaSpoutConfig.Consumer.BOOTSTRAP_SERVERS, ...) props.put(KafkaSpoutConfig.Consumer.GROUP_ID, "kafka-spout") props.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, keyDeserializer) props.put(KafkaSpoutConfig.Consumer.VALUE_DESERIALIZER, valueDeserializer) any hints? was: I've created a topology that will read from kafka using storm-kafka-client but when it reaches the last message on kafka log it stops consuming and get stuck, new messages are never consumed, here are the kafka logs: ``` [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Preparing to restabilize group kafka-spout with old generation 29 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Stabilized group kafka-spout generation 30 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:15:26,868] INFO [GroupCoordinator 1002]: Assignment received from leader for group kafka-spout for generation 30 (kafka.coordinator.GroupCoordinator) ``` ========= here storm starts consuming messages, then, when it hits the last message, I can see this log in kafka == > [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Preparing to restabilize group kafka-spout with old generation 30 (kafka.coordinator.GroupCoordinator) [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Group kafka-spout with generation 31 is now empty (kafka.coordinator.GroupCoordinator) ===== and then storm consumer group is stuck, no new messages are read from kafka. my topology/ spout are configured that way: Topology: c.put(SConfig.TOPOLOGY_MAX_SPOUT_PENDING, 1000) c.put(SConfig.NIMBUS_SEEDS, "my nimbus seeds") c.put(SConfig.NIMBUS_THRIFT_PORT, 6627) c.put(SConfig.TOPOLOGY_WORKERS, 2) c.put(SConfig.TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS, 100) Spout: props.put(KafkaSpoutConfig.Consumer.ENABLE_AUTO_COMMIT, true) props.put(KafkaSpoutConfig.Consumer.BOOTSTRAP_SERVERS, ...) props.put(KafkaSpoutConfig.Consumer.GROUP_ID, "kafka-spout") props.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, keyDeserializer) props.put(KafkaSpoutConfig.Consumer.VALUE_DESERIALIZER, valueDeserializer) any hints? > storm-kafka-client consumer group getting stuck consuming from kafka 0.10.1.1 > ----------------------------------------------------------------------------- > > Key: STORM-2342 > URL: https://issues.apache.org/jira/browse/STORM-2342 > Project: Apache Storm > Issue Type: Bug > Components: storm-kafka-client > Affects Versions: 1.0.2 > Reporter: Felipe Cavalcanti > > I've created a topology that will read from kafka using storm-kafka-client > but when it reaches the last message on kafka log it stops consuming and get > stuck, new messages are never consumed, here are the kafka logs: > [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Preparing to > restabilize group kafka-spout with old generation 29 > (kafka.coordinator.GroupCoordinator) > [2017-02-03 19:15:26,865] INFO [GroupCoordinator 1002]: Stabilized group > kafka-spout generation 30 (kafka.coordinator.GroupCoordinator) > [2017-02-03 19:15:26,868] INFO [GroupCoordinator 1002]: Assignment received > from leader for group kafka-spout for generation 30 > (kafka.coordinator.GroupCoordinator) > ========= here storm starts consuming messages, then, when it hits the last > message, I can see this log in kafka == > > [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Preparing to > restabilize group kafka-spout with old generation 30 > (kafka.coordinator.GroupCoordinator) > [2017-02-03 19:16:01,266] INFO [GroupCoordinator 1002]: Group kafka-spout > with generation 31 is now empty (kafka.coordinator.GroupCoordinator) > ===== > and then storm consumer group is stuck, no new messages are read from kafka. > my topology/ spout are configured that way: > Topology: > c.put(SConfig.TOPOLOGY_MAX_SPOUT_PENDING, 1000) > c.put(SConfig.NIMBUS_SEEDS, "my nimbus seeds") > c.put(SConfig.NIMBUS_THRIFT_PORT, 6627) > c.put(SConfig.TOPOLOGY_WORKERS, 2) > c.put(SConfig.TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS, 100) > Spout: > props.put(KafkaSpoutConfig.Consumer.ENABLE_AUTO_COMMIT, true) > props.put(KafkaSpoutConfig.Consumer.BOOTSTRAP_SERVERS, ...) > props.put(KafkaSpoutConfig.Consumer.GROUP_ID, "kafka-spout") > props.put(KafkaSpoutConfig.Consumer.KEY_DESERIALIZER, keyDeserializer) > props.put(KafkaSpoutConfig.Consumer.VALUE_DESERIALIZER, valueDeserializer) > > any hints? -- This message was sent by Atlassian JIRA (v6.3.15#6346)