Apache Camel 3.21.x Kafka Consumer max.poll.interval.ms类型异常求助

问题场景 #

在Spring Boot项目中使用camel-spring-boot-starter和camel-kafka-starter构建Kafka Consumer，路由配置如下：

from("kafka:my-topic?brokers=myBroker&consumersCount=1&groupId=myGroupId&maxPollRecords=1&maxPollIntervalMs=3600000&autoCommitEnable=false&allowManualCommit=true")

该配置在Camel 3.20.6版本正常运行，但升级到3.21.0/3.21.1版本后，启动抛出异常：

org.apache.kafka.common.config.ConfigException: Invalid value 3600000 for configuration max.poll.interval.ms: Expected value to be a 32-bit integer, but it was a java.lang.Long
    at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:688)
    at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
    at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
    at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:113)
    at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:133)
    at org.apache.kafka.clients.consumer.ConsumerConfig.<init>(ConsumerConfig.java:630)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:664)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:645)
    at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:625)
    at org.apache.camel.component.kafka.DefaultKafkaClientFactory.getConsumer(DefaultKafkaClientFactory.java:34)
    at org.apache.camel.component.kafka.KafkaFetchRecords.createConsumer(KafkaFetchRecords.java:245)
    at org.apache.camel.component.kafka.KafkaFetchRecords.createConsumerTask(KafkaFetchRecords.java:205)
    at org.apache.camel.support.task.ForegroundTask.run(ForegroundTask.java:93)
    at org.apache.camel.component.kafka.KafkaFetchRecords.run(KafkaFetchRecords.java:127)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)

尝试传入更小数值（如360）仍报错，直接传字符串形式"3600000"会被解析为0。

原因分析 #

Camel 3.21.x版本对Kafka组件的参数解析逻辑发生了变化，将原本应该转换为Integer类型的参数（如maxPollIntervalMs）解析成了Long类型。而Kafka客户端的ConfigDef要求max.poll.interval.ms必须是32位整数，类型不匹配导致抛出配置异常。

解决方案 #

1. 在路由URI中显式转换参数类型 #

使用Camel支持的SpEL表达式，将参数值强制转换为Integer类型，替换原路由中的maxPollIntervalMs参数：

from("kafka:my-topic?brokers=myBroker&consumersCount=1&groupId=myGroupId&maxPollRecords=1&maxPollIntervalMs=#{'3600000' as Integer}&autoCommitEnable=false&allowManualCommit=true")

2. 改用Spring Boot配置文件统一配置 #

将Kafka组件的参数移到Spring Boot配置文件中，避免URI中的类型解析问题：

• application.yml：

camel:
  component:
    kafka:
      configuration:
        brokers: myBroker
        consumers-count: 1
        group-id: myGroupId
        max-poll-records: 1
        max-poll-interval-ms: 3600000
        auto-commit-enable: false
        allow-manual-commit: true

• 路由简化为：

from("kafka:my-topic")

3. 临时降级到Camel 3.20.6版本 #

如果上述方案无法快速验证，可暂时降级回Camel 3.20.6版本，等待官方修复该类型解析的bug。

内容的提问来源于stack exchange，提问作者Alexxxx