если установить spring.kafka.listener.ack-mode = время, будет ли работать снова?или повторить работу в указанном ack-режиме? - PullRequest
Новая
       1

если установить spring.kafka.listener.ack-mode = время, будет ли работать снова?или повторить работу в указанном ack-режиме?

0 голосов
/ 25 февраля 2019

Spring Kafka - Как повторить попытку с @ KafkaListener

spring-kafka SeekToCurrentErrorHandler, если установлено spring.kafka.listener.ack-mode=time, будет ли повторная попытка работать?или повторите работу в указанном режиме подтверждения, например, MANUAL MANUAL_IMMEDIATE. 

@Value("${${sync.kafka.header.source.id}}")
private String headerSourceId;

@Value("${sync.kafka.from.id}")
private String syncKafkaFromId;

@Value("${sync.kafka.to.id}")
private String syncKafkaToId;

@Autowired
private KafkaTemplate kafkaTemplate;

@Autowired
private ConsumerFactory consumerFactory;

@KafkaListener(topics = "${sync.kafka.topics}")
public void listen(ConsumerRecord<?, ?> record,
                   @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                   @Header(value = KafkaHeaders.RECEIVED_MESSAGE_KEY, required = false) String messageKey,
                   @Header(value = "${sync.kafka.header.source.id}", required = false) String customHeader,
                   @Headers MessageHeaders messageHeaders) {
    log.info("- - - - - - - - - - - - - - -");
    log.info("topic: {}", topic);
    log.info("message key: {}", messageKey);
    log.info("custom header: {}", customHeader);
    log.info("messageHeaders: {}", messageHeaders);

    ProducerRecord<Object, Object> ProducerRecord = new ProducerRecord(topic,record.key(), record.value());
    ProducerRecord.headers().add(new RecordHeader(headerSourceId, syncKafkaFromId.getBytes()));

    kafkaTemplate.send(ProducerRecord);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
    // or factory.setRetryTemplate(aRetryTemplate);
    // and factory.setRecoveryCallback(aRecoveryCallback);
    return factory;
}

1 Ответ

0 голосов
/ 25 февраля 2019

При использовании SeekToCurrentErrorHandler вы должны использовать только ручной, запись или пакетный режим подтверждения.Вам также следует установить свойство контейнера ackOnError на false и установить enable.auto.commit на false.

...