приложение работает, но через некоторое время перестает потреблять сообщения - PullRequest
Новая
       13

приложение работает, но через некоторое время перестает потреблять сообщения

0 голосов
/ 08 мая 2019

Spring-kafka потребитель перестает потреблять сообщения через некоторое время.Остановка происходит каждый раз, но никогда не в одно и то же время.Когда приложение больше не потребляет, в конце журнала всегда вижу утверждение, что потребитель отправил сигнал LEAVE_GROUP.Если я не вижу никаких ошибок или исключений, почему потребитель покидает группу? 

org.springframework.boot:spring-boot-starter-parent:2.0.4.RELEASE
spring-kafka:2.1.8.RELEASE
org.apache.kafka:kafka-clients:1.0.2

Я установил логирование как logging.level.org.apache.kafka = DEBUG logging.level.org.springframework.kafka = INFO

другие настройки 

spring.kafka.listener.concurrency=5
spring.kafka.listener.type=single
spring.kafka.listener.ack-mode=record
spring.kafka.listener.poll-timeout=10000
spring.kafka.consumer.heartbeat-interval=5000
spring.kafka.consumer.max-poll-records=50
spring.kafka.consumer.fetch-max-wait=10000
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.properties.security.protocol=SSL
spring.kafka.consumer.retry.maxAttempts=3
spring.kafka.consumer.retry.backoffperiod.millisecs=2000

Настройка ContainerFactory 

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> recordsKafkaListenerContainerFactory(RetryTemplate retryTemplate) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(listenerCount);
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.RECORD);
        factory.getContainerProperties().setPollTimeout(pollTimeoutMillis);
        factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
        factory.getContainerProperties().setAckOnError(false);
        factory.setRetryTemplate(retryTemplate);
        factory.setStatefulRetry(true);
        factory.getContainerProperties().setIdleEventInterval(60000L);


        return factory;
    }

Настройка прослушивателя 

@Component
public class RecordsEventListener implements ConsumerSeekAware {

    private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(RecordsEventListener.class);



    @Value("${mode.replay:false}")
    public void setModeReplay(boolean enabled) {

        this.isReplay = enabled;
    }

    @KafkaListener(topics = "${event.topic}", containerFactory = "RecordsKafkaListenerContainerFactory")
    public void handleEvent(@Payload String payload) throws RecordsEventListenerException {

        try {
           //business logic
        } catch (Exception e) {

            LOG.error("Process error for event: {}",payload,e);

            if(isRetryableException(e)) {
                LOG.warn("Retryable exception detected. Going to retry.");
                throw new RecordsEventListenerException(e);
            }else{
                LOG.warn("Dropping event because non retryable exception");
            }
        }

    }

    private Boolean isRetryableException(Exception e) {

        return binaryExceptionClassifier.classify(e);
    }





    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        //do nothing
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {

        //do this only once per start of app
        if (isReplay && !partitonSeekToBeginningDone) {
            assignments.forEach((t, p) -> callback.seekToBeginning(t.topic(), t.partition()));
            partitonSeekToBeginningDone = true;
        }

    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        //do nothing
        LOG.info("Container is IDLE; no messages to pull.");
        assignments.forEach((t,p)->LOG.info("Topic:{}, Partition:{}, Offset:{}",t.topic(),t.partition(),p));
    }

    boolean isPartitionSeekToBeginningDone() {

        return partitonSeekToBeginningDone;
    }

    void setPartitonSeekToBeginningDone(boolean partitonSeekToBeginningDone) {

        this.partitonSeekToBeginningDone = partitonSeekToBeginningDone;
    }
}

Когда приложение больше не потребляетВ конце журнала всегда вижу утверждение, что потребитель отправил сигнал LEAVE_GROUP. 

2019-05-02 18:31:05.770 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Sending Heartbeat request to coordinator x.x.x.com:9093 (id: 2147482638 rack: null)
2019-05-02 18:31:05.770 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=app] Using older server API v0 to send HEARTBEAT {group_id=app,generation_id=6,member_id=consumer-1-98d28e69-b0b9-4c2b-82cd-731e53b74b87} with correlation id 5347 to node 2147482638
2019-05-02 18:31:05.872 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Received successful Heartbeat response
2019-05-02 18:31:10.856 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Sending Heartbeat request to coordinator x.x.x.com:9093 (id: 2147482638 rack: null)
2019-05-02 18:31:10.857 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=app] Using older server API v0 to send HEARTBEAT {group_id=app,generation_id=6,member_id=consumer-1-98d28e69-b0b9-4c2b-82cd-731e53b74b87} with correlation id 5348 to node 2147482638
2019-05-02 18:31:10.958 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Received successful Heartbeat response
2019-05-02 18:31:11.767 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Sending LeaveGroup request to coordinator x.x.x.com:9093 (id: 2147482638 rack: null)
2019-05-02 18:31:11.767 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-1, groupId=app] Using older server API v0 to send LEAVE_GROUP {group_id=app,member_id=consumer-1-98d28e69-b0b9-4c2b-82cd-731e53b74b87} with correlation id 5349 to node 2147482638
2019-05-02 18:31:11.768 DEBUG 9548 --- [kafka-coordinator-heartbeat-thread | app] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=app] Disabling heartbeat thread

Полный журнал

1 Ответ

0 голосов
/ 14 июня 2019

Спасибо всем, кто ответил. Оказывается, это был действительно брокер, выбрасывающий потребителя по тайм-ауту сессии. Брокер в очень старой версии (0.10.0.1) не приспособлен к новым функциям, описанным в KIP-62 , которую могла использовать используемая нами версия spring-kafka. Поскольку мы не могли диктовать обновление брокеру или изменения времени ожидания сеанса, мы просто изменили нашу логику обработки, чтобы завершить задачу в течение времени ожидания сеанса.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...