Как создать приемник Kafka для потребителей с пружинным запуском, который в случае отклонения сообщения попытается использовать их через переменное время - PullRequest
Новая
       32

Как создать приемник Kafka для потребителей с пружинным запуском, который в случае отклонения сообщения попытается использовать их через переменное время

1 голос
/ 11 июля 2019

У меня есть простой слушатель kafka в приложении Springboot, например: 

@KafkaListener(topics="mytopic")
public void receive(String message) {
   LOGGER.info("received message='{}'", messge);
}

в некоторых особых случаях я хотел бы отклонить сообщение, но я бы хотел, чтобы система предложила мне его снова через определенное время;

как мне это сделать?

Примечание: я также хотел бы, чтобы конфигурация kafka была сделана на заказ (не по умолчанию для Springboot)

Ответы [ 2 ]

1 голос
/ 11 июля 2019

моя реализация выполняет только то, что вам нужно:

1) класс конфигурации kafka, который берет поля из пользовательского свойства и повторяет отклоненные сообщения через 5000 миллисекунд (внутри метода kafkaListenerContainerFactory): 

@Configuration
public class KafkaConfig {

    //...

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        if(enableSsl) {
            //configure the following three settings for SSL Encryption
            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,  sslPassword);

            // configure the following three settings for SSL Authentication
            props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, sslKeystoreLocation);
            props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, sslPassword);
            props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, sslPassword);
        }
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {              
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));

        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);

        RetryTemplate retryTemplate = new RetryTemplate();
        factory.setStatefulRetry(false);
        factory.setRetryTemplate(retryTemplate);

        //infinite number of retry attempts
        retryTemplate.setRetryPolicy(new AlwaysRetryPolicy());

        //wait a "waitingTime" time before retrying
        int waitingTime = 5000;
        FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
        fixedBackOffPolicy.setBackOffPeriod(waitingTime);
        retryTemplate.setBackOffPolicy(fixedBackOffPolicy);

        //or use exponential waiting
        //ExponentialBackOffPolicy expBackoff = new ExponentialBackOffPolicy();
        //expBackoff.setInitialInterval(...);
        //expBackoff.setMaxInterval(...);
        //retryTemplate.setBackOffPolicy(expBackoff);

        return factory;
    }
}

2) класс, который потребляет сообщения: 

@Service
public class Consumer {

    private final Logger logger = LoggerFactory.getLogger(Consumer.class);

    //...

    @KafkaListener(topics="${kafka.topics.test}")
    public void consume(String message, Acknowledgment ack) throws IOException {
        if(processMessage) {
            logger.info(String.format("##KAFKA## -> Consumed message -> %s", message)); 
            ack.acknowledge();
        } 
        else { 
            logger.error(String.format("##KAFKA## -> Failed message -> %s", message));  
            throw new IOException("reject message");
        }
    }
}
1 голос
/ 11 июля 2019

См. повторные доставки и Повторная проверка с отслеживанием состояния .

Сконфигурируйте фабрику слушателя с желаемыми характеристиками повторения и (опционально) добавьте SeekToCurrentErrorHandler.

...