Spring Boot 3迁移后Spring Cloud Stream Kinesis无法接收消息求助

以下是针对你场景的具体排查和修复步骤：

1. 确认函数式绑定配置的正确性 #

你的消费者Bean是Consumer<String> authorization，对应Spring Cloud Stream函数式模型的输入绑定名称为authorization-in-0，需确保配置文件中存在正确的绑定映射：

# 指定要激活的函数（必填，否则Consumer Bean不会被绑定到流）
spring.cloud.stream.function.definition=authorization

# 绑定函数到目标Kinesis流
spring.cloud.stream.bindings.authorization-in-0.destination=你的Kinesis流名称
spring.cloud.stream.bindings.authorization-in-0.group=你的消费者组名

注意：Spring Cloud Stream 4.x（配合Spring Boot 3）默认启用函数式模型，必须显式指定spring.cloud.stream.function.definition，否则不会自动关联你的Consumer Bean。

2. 检查Kinesis消费者初始位置配置 #

如果流中已有历史消息但无法接收，可能是默认初始位置为LATEST（只接收配置后新产生的消息），需调整为TRIM_HORIZON以消费历史消息：

spring.cloud.stream.kinesis.bindings.authorization-in-0.consumer.starting-position=TRIM_HORIZON

3. 验证IAM权限与Kinesis流配置 #

确保应用使用的IAM角色具备以下Kinesis权限：

• kinesis:DescribeStream
• kinesis:GetRecords
• kinesis:GetShardIterator
• kinesis:ListShards
  同时确认Kinesis流的区域与应用配置的AWS区域一致：

spring.cloud.aws.region.static=你的AWS区域（如us-east-1）

4. 排查依赖冲突与版本兼容性 #

虽然你已指定正确版本，但需确保没有引入旧版本的Spring Cloud AWS或Spring Cloud Stream依赖。可以通过mvn dependency:tree（Maven）或gradle dependencies（Gradle）检查依赖树，排除任何旧版本的冲突依赖，例如：

<!-- 如果存在冲突，在spring-cloud-starter-stream-kinesis依赖中排除旧AWS模块 -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kinesis</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-aws-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>

5. 开启Debug日志定位问题 #

添加日志配置，查看Kinesis Binder的详细运行日志，确认是否有连接、分片订阅或消息拉取的静默异常：

logging.level.org.springframework.cloud.stream=DEBUG
logging.level.org.springframework.cloud.aws.kinesis=DEBUG
logging.level.com.amazonaws.services.kinesis=DEBUG

重点关注日志中是否有ShardConsumer相关的启动信息，以及是否有拉取记录的日志输出。

6. 确认消费者组的正确性 #

如果同一流存在多个消费者组，确保你的应用使用的消费者组未被其他进程占用，或者配置了正确的分片分配策略：

spring.cloud.stream.kinesis.bindings.authorization-in-0.consumer.shard-offset-initializer=org.springframework.cloud.stream.binder.kinesis.config.KinesisConsumerProperties$TrimHorizonShardOffsetInitializer

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