我有一个java使用者，它不断地运行着大量的数据和消息。。有时，程序会被卡在消费者身上。。它看起来像死锁或无限循环，但我不确定。没有抛出异常，程序似乎工作正常，尽管它不轮询实际数据(这个问题只发生在java程序中。。我运行console consumer，效果很好）
唯一的办法就是重新启动程序。。。

public class Main {
Properties consumerComfig = new Properties ();
consumerConfig.put(ConsumerConfig. KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"Org.apache.kafka.comnect.json.JsonDeserializer");
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "some server")
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID());
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"):
consumerConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
consumerConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,10000)
consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"):
consumer = new KafkaConsumer<String, JsonNode> (consumerConfig);

public void consumeAndSendToCorrelator () {
    logger.info ("Starting kafka consumer, listening to topics stringBuilder.toString ()");
    try {
        consumer.subscribe("INPUT TOPICS");
        while (true) {
            ConsumerRecords<String, JsonNode> records = consumer.poll(8000);
            logger.debug("Trying to pull records from consumer");
            JsonNode value = null;
            for (Consumer Record<String, JsonNode > irecord :records){
                try {
                    value = record.value();
                    if (value == null)
                        continue;
                    //Somme thread that process the message
                    correlatorActor.tell(value, ActorRef.noSender);
                } catch (JsonProcessingException e) {
                    logger.error("Could not parse incoming json" + value.toString());
                } catch (IOException e) {
                    logger.error("IOException occurred", e);
                } catch (Exception e) {
                    logger.error("Error occurred", e);
                }
                consumer.commitSync();
            }
        }
    }catch(WakeupException e) {
        logger.info("closed consumer");
    }finally {
        consumer.close()
    }
}

}
我用的是Kafka0.10你知道会发生什么吗？我能做些什么来修复它？