Uploaded image for project: 'Kafka'
  1. Kafka
  2. KAFKA-7378

Consumer poll hangs if broker shutdown while consumer attempting to connect.

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Open
    • Priority: Major
    • Resolution: Unresolved
    • Affects Version/s: 0.11.0.1
    • Fix Version/s: None
    • Component/s: consumer
    • Labels:
      None
    • Environment:
      Java JDK 8, Linux RedHat 7.5

      Description

      I have an integration test that creates a Spring context, starting up the application. In the test's @Before method, I create a zookeeper and KafkaServer instance with available ephemeral service ports. The test's @After method shuts down these service instances. The Spring context registers a shutdown handler, which doesn't get called till after the @After method is run. This means my application's consumer polling loop doesn't detect a termination event until after the broker is gone. Here's what shows up in my log4j log:

       

      2018-09-05 11:56:13.980,1819313073756610 {} DEBUG o.a.k.c.NetworkClient [hmdb-kafka-polling-thread] Initiating connection to node localhost:9092 (id: -1 rack: null)
      2018-09-05 11:56:13.980,1819313074091635 {} DEBUG o.a.k.c.n.Selector [hmdb-kafka-polling-thread] Connection with localhost/127.0.0.1 disconnected
      java.net.ConnectException: Connection refused
      at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[?:1.8.0_171]
      at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) ~[?:1.8.0_171]
      at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:50) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:95) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:361) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.common.network.Selector.poll(Selector.java:326) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:433) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:232) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:208) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:199) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:226) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:203) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:286) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078) ~[kafka-clients-0.11.0.1.jar:?]
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043) ~[kafka-clients-0.11.0.1.jar:?]
      

       

      This sequence continues forever because poll will not return and notice it's time to terminate my polling loop.

        Attachments

          Activity

            People

            • Assignee:
              Unassigned
              Reporter:
              john.calcote@gmail.com John Calcote
            • Votes:
              0 Vote for this issue
              Watchers:
              1 Start watching this issue

              Dates

              • Created:
                Updated: