监控 Kafka Consumer Lag 并生成警报

Question

如何监控 Kafka 消费者延迟并生成电子邮件/警报？以下是我的要求

1. 我想在主题超过 1 天的消息时触发电子邮件。

我正在使用 Spring Boot 微服务，Java 8

@Configuration
public class KafkaConsumerConfig 
{
    @Value(value = "${kafka.bootstrapAddress}")
    private String bootstrapAddress;
 
    @Value(value = "${general.topic.group.id}")
    private String groupId;
 
    @Value(value = "${user.topic.group.id}")
    private String userGroupId;
 
    // 1. Consume string data from Kafka
 
    @Bean
    public ConsumerFactory<Integer, String>  consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                StringDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        return new DefaultKafkaConsumerFactory<>(props);
    }
    
    
 
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
    
 //not compiling
    public KafkaMessageListenerContainer<Integer, String> m1()
    {
        ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
        containerProps.setMessageListener(new MessageListener<Integer, String>() { 
 
            @Override
            public void onMessage(ConsumerRecord<Integer, String> data) {
                // TODO Auto-generated method stub
                
            }
           
        });
        DefaultKafkaConsumerFactory<Integer, String> cf =
                                new DefaultKafkaConsumerFactory<>(consumerFactory()); //not compiling
        KafkaMessageListenerContainer<Integer, String> container =
                                new KafkaMessageListenerContainer<>(cf, containerProps);
        return container;
    }

Answer 1

您可以使用Burrow 而不是尝试编写自己的解决方案。