Spring集成消息驱动通道适配器不能与Spring-Kafka 2.3+一起工作

Question

试图调用不存在的方法。尝试是从以下位置进行的：

org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.onInit(KafkaMessageDrivenChannelAdapter.java:318)

以下方法不存在：

org.springframework.kafka.listener.ContainerProperties.isDeliveryAttemptHeader()Z

3.如果您使用kafka版本2.5及以上版本，但却被2021-03-22 13:56:05.102-0400 org{local_sparta}WARN[data-pipeline,,,][DP-ACCOUNT][DPA][]annotationConfigServletWebServerApplicationContext：上下文初始化过程中遇到的主要异常-取消刷新尝试：org.springframework.beans.factory.beanCreationException：创建名为嵌套异常是org.springframework.beans.beanInstantiationException：无法实例化[org.springframework.boot.autoconfigure.kafka.kafkaAnnotationDrivenConfiguration]:构造函数引发异常；嵌套异常是org.springframework.beans.factory.unsatisfieddependencyexception：创建类路径资源[org/springframework/boot/autoconfigure/kafka/kafkaautoconfiguration.class]中定义的名为“kafka template”的bean时出错：通过方法“kafka template”参数0表示不满足的依赖关系；嵌套异常是org.springframework.beans.factory.NosuchBeanDefinitionException：没有类型为'org.springframework.kafka.core.producerFactory '的合格bean可用：至少需要1个符合autowire候选的bean。依赖项批注：{}

@Configuration
@Slf4j
public class KafkaChannelConsumer {

    @Autowired
    MessageChannel preRouterLOB;

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

    @Value("${spring.kafka.topic:55iptest}")
    private String springIntegrationKafkaTopic;

    @Bean
    public KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter() {
        KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter(
                kafkaListenerContainer());
        kafkaMessageDrivenChannelAdapter.setOutputChannel(preRouterLOB);
        return kafkaMessageDrivenChannelAdapter;
    }

    @SuppressWarnings("unchecked")
    @Bean
    public ConcurrentMessageListenerContainer kafkaListenerContainer() {
        ContainerProperties containerProps = new ContainerProperties(springIntegrationKafkaTopic);

        return (ConcurrentMessageListenerContainer) new ConcurrentMessageListenerContainer(
                consumerFactory(), containerProps);
    }

    @Bean
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory(consumerConfigs());
    }

    @Bean
    public Map consumerConfigs() {
        Map properties = new HashMap();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "dummy");
        return properties;
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
  xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jms="http://www.springframework.org/schema/integration/jms"
       xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
       xmlns:int="http://www.springframework.org/schema/integration"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">


<int-kafka:message-driven-channel-adapter
        id="kafkaListener"
        listener-container="container1"
        auto-startup="false"
        phase="100"
        send-timeout="5000"
        mode="record"
        channel="someChannel"
        error-channel="errorChannel" />

<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
    <constructor-arg>
        <bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
            <constructor-arg>
                <map>
                <entry key="bootstrap.servers" value="localhost:9092" />
                </map>
            </constructor-arg>
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.kafka.listener.config.ContainerProperties">
            <constructor-arg name="topics" value="foo" />
        </bean>
    </constructor-arg>

</bean>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-kafka</artifactId>
            <version>5.4.5</version>
        </dependency>

这包括Spring-Kafka 2.3.6版本

第3期POM

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-kafka</artifactId>
    <version>5.4.5</version>
    <exclusions>
       <exclusion>
           <groupId>org.springframework.kafka</groupId>
           <artifactId>spring-kafka</artifactId>
       </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.6.7</version>
</dependency>
       

Answer 1

5.4.5

这包括Spring-Kafka 2.3.6版本

不，它没有；spring-integration-kafka的5.4.x版本需要2.6.x；该方法被添加到2.5中的属性中。