title: Apache kafka consumer poll infinite loop on unexpected messages
body: <p>I'm doing integration tests of Kafka consumer implementation.
  I use wurstmeister/kafka docker image and Apache Kafka consumer.
  The scenario that buzzes me is when I send "unexpected" messages to a topic. The <code>kafkaConsumer.poll(POLLING_TIMEOUT)</code> seems to go in an infinite loop in RUN mode. When I DEBUG though, it works when I pause and run back.</p>
  
  <p>I don't have this issue when sending messages that are expected ( do not throw an exception on deserialization).</p>
  
  <p>Here is my <code>docker-compose</code> configuration for kafka :</p>
  
  <pre><code>kafka:
    image: wurstmeister/kafka
    links:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: localhost
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_CREATE_TOPICS: "ProductLocation:1:1,ProductInformation:1:1,InventoryAvailableToSell:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
  </code></pre>
  
  <p>The Java generic consumer :</p>
  
  <pre><code>@Override
  public Collection&lt;T&gt; consume() {
      String eventToBePublishedName = ERROR_WHILE_RESETTING_OFFSET;
      boolean success = false;
  
      try {
          kafkaConsumer.resume(kafkaAssignments);
          if (isPollingTypeFull) {
              // dummy poll because its needed before resetting offset.
              // https://stackoverflow.com/questions/41008610/kafkaconsumer-0-10-java-api-error-message-no-current-assignment-for-partition
              kafkaConsumer.poll(POLLING_TIMEOUT);
              resetOffset();
          } else if (!offsetGotResetFirstTime) {
              resetOffset();
              offsetGotResetFirstTime = true;
          }
  
          eventToBePublishedName = ERROR_WHILE_POLLING;
  
          ConsumerRecords&lt;Object, T&gt; records;
  
          List&lt;T&gt; output = new ArrayList&lt;&gt;();
  
          do {
              records = kafkaConsumer.poll(POLLING_TIMEOUT);
              records.forEach(cr -&gt; {
                  T val = cr.value();
                  if (val != null) {
                      output.add(cr.value());
                  }
              });
          } while (records.count() &gt; 0);
  
          eventToBePublishedName = CONSUMING;
          success = true;
          kafkaConsumer.pause(kafkaAssignments);
          return output;
      } finally {
          applicationEventPublisher.publishEvent(
                  new OperationResultApplicationEvent(
                          this, OperationType.ConsumingOfMessages, eventToBePublishedName, success));
      }
  }
  </code></pre>
  
  <p>The deserializing :</p>
  
  <pre><code>public T deserialize(String topic, byte[] data) {
      try {
          JsonNode jsonNode = mapper.readTree(data);
          JavaType javaType = mapper.getTypeFactory().constructType(getValueClass());
          JsonNode value = jsonNode.get("value");
          return mapper.readValue(value.toString(), javaType);
      } catch (IllegalArgumentException | IOException | SerializationException e) {
          LOGGER.error("Can't deserialize data [" + Arrays.toString(data)
                  + "] from topic [" + topic + "]", e);
          return null;
      }
  }
  </code></pre>
  
  <p>In my integration tests, I create a topic for every test by sending to a time-stamped topic name. This creates new topics and makes tests stateless.</p>
  
  <p>This is how I configure the Kafka consumer :</p>
  
  <pre><code>Properties properties = new Properties();
      properties.put("bootstrap.servers", kafkaConfiguration.getServer());
      properties.put("group.id", kafkaConfiguration.getGroupId());
      properties.put("key.deserializer", kafkaConfiguration.getKeyDeserializer().getName());
      properties.put("value.deserializer", kafkaConfiguration.getValueDeserializer().getName());
  </code></pre>