Слушатели фабрики Kafka 2 постоянно не подключены

Question

Мы используем версию Spring Kafka 2.1.4.RELEASE в нашем проекте, и у нас есть следующая конфигурация:

@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Configuration
    class ProducerConfig {
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ASerializer.class);
            return props;
        }

        @Bean
        public ProducerFactory<String, A> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public KafkaTemplate<String, A> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }

    @Configuration
    class ConsumerConfig {
        @Value("${spring.kafka.consumer.group-id}")
        private String groupId;
        @Value("${spring.kafka.consumer.auto-offset-reset}")
        private String autoOffsetReset;
        @Value("${spring.kafka.consumer.enable-auto-commit}")
        private boolean enableAutoCommit;
        @Value("${spring.kafka.consumer.max-poll-records}")
        private Integer maxPollRecords;

        @Bean
        public Map<String, Object> firstConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ADeserializer.class);
            return props;
        }

        @Bean
        public Map<String, Object> secondConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BDeserializer.class);
            return props;
        }

        @Bean
        public ConsumerFactory<String, A> firstConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(firstConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, A>> firsttContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, A> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(firstConsumerFactory());
            return factory;
        }

        @Bean
        public ConsumerFactory<String, B> secondConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(secondConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, B>> outputTopicContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, B> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(secondConsumerFactory());
            return factory;
        }


        private Map<String, Object> getCommonConsumerConfig() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
            return props;
        }
    }
}

Итак, как вы можете видеть, мы используем 2 фабрики потребителей.И у нас есть следующий потребительский класс:

@ConfigurationProperties
@Service
public class Listener {

    @KafkaListener(containerFactory = "firstContainerFactory", topics = "someTopic")
    public void firstListener(@Payload A first) {
        //some logic
    }

    @KafkaListener(containerFactory = "secondContainerFactory", topics = {
     //topic list       
    })
    public void secondTopicListener(@Payload B second) {
       //some logic

    }


}

Итак, при запуске этого приложения мы замечаем, что оно не связано с обеими темами постоянно.Иногда это связано только со второй темой или только с первой темой, и, возможно, это может быть связано с первой и второй темами (что правильно).Так что вы можете помочь понять, что здесь неправильно настроено?

Answer 1

Хорошо, после еще одного расследования я смог определить, какие проблемы произошли на моей стороне.Так что в основном у нас есть одна группа потребителей с несколькими темами.Так что в моем случае у нас было 0 разделов для каждой из этих тем (насколько я понимаю, разделов нет, и мы работаем с 1 очередью темы).Поэтому, когда я подключаюсь к этому экземпляру kafka - все потребители подключаются к этим темам, но когда кто-то также подключается к этим темам (возможно, мой коллега), происходит перебалансирование, он начинает слушать одну из этих тем вместо меня (из-зафакт, что только один потребитель может представить на раздел).

Answer 2

Как правило, рекомендуется размещать каждого слушателя в отдельном group.id (вы можете использовать атрибут groupId в @KafkaListener, который переопределяет фабрику потребителей).В противном случае, когда начинается вторая, на первый возникает перебалансировка.Текущая версия 2.1.x - 2.1.10.