Кафка опрос внезапно застрял - PullRequest
Новая
       11

Кафка опрос внезапно застрял

2 голосов
/ 02 июля 2019

У меня есть потребитель Java, который постоянно работает с большим количеством данных и сообщений .. иногда программа зависает на получателе данных. Это похоже на тупик или бесконечный цикл, но я не уверен в этом. Не возникает никаких исключений, и программа, кажется, работает должным образом, хотя она не опрашивает фактические данные. (Проблема возникает только в Java-программе. Я запускаю консольный потребитель, и он отлично работает)

Единственный способ это исправить - перезапустить программу ... 

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()
    }
}

}

Я использую Кафку 0.10 Вы знаете, что с этим происходит? Что я могу сделать, чтобы это исправить?

...