[KAFKA-8401] consumer.poll(Duration.ofMillis(100)) blocking

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.1.0
• Component/s: consumer
• Labels: blocking kafka
• Environment: kafka 1.1.0 ; zk 3.4.11
• Flags: Patch
• Assignee: Unassigned
• Reporter: leishuiyu
• Created: 21/May/19 11:25
• Updated: 21/May/19 11:27

Description

1. this is code

   public class Consumer extends Thread {
   
       KafkaConsumer<Integer, String> consumer;
   
       public Consumer() {
           Properties props = new Properties();
           //47.105.201.137 is public network Ip
           props.put("bootstrap.servers", "47.105.201.137:9092");  //连接地址
           props.put("group.id", "lsy_test");
           props.put("zookeeper.session.timeout.ms", "400");
           props.put("zookeeper.sync.time.ms", "200");
           props.put("auto.commit.interval.ms", "1000");
           props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
           props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
           this.consumer = new KafkaConsumer<Integer, String>(props);
       }
   
   
       @Override
       public void run() {
           consumer.subscribe(Arrays.asList("flink_order"));
           while (true) {
               ConsumerRecords<Integer, String> poll = consumer.poll(Duration.ofMillis(100));
               for (ConsumerRecord<Integer, String> record : poll) {
                   System.out.println(record.key() + "-------" + record.value());
               }
           }
       }
   
       public static void main(String[] args) {
           Consumer sumer = new Consumer();
           sumer.start();
       }
   }
   
   

1.  Configured hosts for remote machines

   xx.xx.xx.xx centos-7

1. when my code running in local machines,the bootstrap.servers=47.105.201.137:9092 the consumer poll is blocking ,howerver in my mac set /etc/hosts 47.105.201.137 centos-7 and boostrap.servers=centos-7:9092 the consumer can poll message,The previous methods consumer.listTopics() is successful，only poll message is blocking ,I feel very confused