Пружина Кафки Неоднозначные методы для типа полезной нагрузки

Question

Я пытаюсь использовать два различных типа полезных нагрузок из темы Kafka, используя spring-kafka-2.2.0 и spring-boot-2.1.0, но, к сожалению, пока не получилось, исключение.

Так что у меня есть две модели в моем проектеcom.kafka.model.Professor и com.kafka.model.Student, и я также настроил их на отображение типов.Но я не уверен, какая часть мне не хватает в этой десериализации.

Config

@Configuration
public class KafkaConsumerConfig {

@Value("${kafka.consumer.bootstrap-server}")
private String bootstrapServer;

@Bean("consumerConfigs")
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-4");
    props.put(JsonDeserializer.TYPE_MAPPINGS,
            "professor:com.kafka.model.Professor, student:com.kafka.model.Student");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "test-4");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
    return props;
}

@Bean("consumerFactory")
public ConsumerFactory<String, Object> consumerFactory() {
    JsonDeserializer<Object> jsonDeserializer = new JsonDeserializer<>(Object.class);
    jsonDeserializer.addTrustedPackages("*");

    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), jsonDeserializer);
}

@Bean("kafkaListenerContainerFactory")
@ConditionalOnMissingBean(name = "KafkaAutoConfiguration")
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.BATCH);
    return factory;
    }

}

Класс слушателя

@Service
@KafkaListener(topics = "test-events",id = "kafkaListenerContainerFactory-11")
public class KafkaConsumerService {

@KafkaHandler
public void student(List<Student> stu) {
    System.out.println(stu);
}

@KafkaHandler
public void professor(List<Professor> pro) {
    System.out.println(pro);
   }


}

Ошибка

2019-02-15 13:04:33.060 ERROR 14863 --- [actory-11-0-C-1] o.s.k.listener.BatchLoggingErrorHandler  : Error while processing:
ConsumerRecord(topic = stores-pricing-easytest-manual-events, partition = 4, offset = 1189400, CreateTime = 1550096019845, serialized key size = 7, serialized value size = 49, headers = RecordHeaders(headers = [], isReadOnly = false), key = student, value = {studentName=thomas, rollNum=109, age=35})

 org.springframework.kafka.KafkaException: Ambiguous methods for payload type: class java.util.ArrayList: student and professor

Answer 1

Подход @KafkaHandler не работает с режимом BATCH.По крайней мере, так, как у вас сейчас: просто нет способа обработать выделенную партию с помощью общего метода.

Попробуйте отключить пакетный режим или просто обработать все в одном @KafkaListener методе, которыйуже принимает List в качестве аргумента для полезной нагрузки с любой общей информацией.

ПРИМЕЧАНИЕ. Apache Kafka не распределяет десериализованную запись между различными пакетами.Все они будут созданы для слушателя в одном и том же ConsumerRecords.Итак, ваша логика с попыткой распределения по универсальному типу в списке никогда не произойдет, даже если бы мы выяснили универсальный тип аргумента метода ... Итак, простой RECORD more - это ваша заставка.