Spring Boot集成Spring Integration AWS Kinesis启动重复报错求助

Hey there, let's tackle this Kinesis integration issue step by step!

First, Let's Contextualize the Error #

The truncated log snippet you shared (2018-03-21 19:38:20.763 INFO 8417 --- [is-dispatcher-1] a.i.k.KinesisMessageDri...) suggests repeated retries or failures from the Kinesis message driver. Common triggers for this include permission issues, missing stream resources, misconfigured consumer settings, or dependency version conflicts. Let's break down the fixes.

Step 1: Validate Your Maven pom.xml #

First, ensure your dependencies are compatible with your Spring Boot version (for 2018's Spring Boot 2.0.x, use spring-integration-aws 2.0.x). Here's a correct pom snippet:

<dependencies>
    <!-- Core Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <!-- Spring Integration AWS Kinesis -->
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-aws</artifactId>
        <version>2.0.1.RELEASE</version> <!-- Match your Spring Boot version -->
    </dependency>

    <!-- AWS SDK for Kinesis -->
    <dependency>
        <groupId>com.amazonaws</groupId>
        <artifactId>aws-java-sdk-kinesis</artifactId>
        <version>1.11.300</version> <!-- Compatible with Spring Integration AWS 2.0.x -->
    </dependency>
</dependencies>

Pro tip: Always cross-check version compatibility between Spring Boot and Spring Integration AWS — mismatched versions often cause silent retry loops.

Step 2: Fix Core Kinesis Configuration #

Most repeated errors stem from misconfigured access or consumer settings. Here's a working configuration example to reference:

application.properties #

# AWS Credentials (use IAM roles if running on AWS EC2/EKS instead)
cloud.aws.credentials.access-key=YOUR_ACCESS_KEY
cloud.aws.credentials.secret-key=YOUR_SECRET_KEY
cloud.aws.region.static=us-east-1

# Kinesis Stream Details
kinesis.stream.name=your-test-stream
kinesis.consumer.group=my-app-consumer-group

Configuration Class #

@Configuration
@EnableIntegration
public class KinesisIntegrationConfig {

    @Value("${cloud.aws.region.static}")
    private String awsRegion;

    @Value("${kinesis.stream.name}")
    private String streamName;

    @Value("${kinesis.consumer.group}")
    private String consumerGroup;

    @Bean
    public AmazonKinesisAsync amazonKinesisAsync() {
        return AmazonKinesisAsyncClientBuilder.standard()
                .withRegion(awsRegion)
                .build();
    }

    @Bean
    public KinesisMessageDrivenChannelAdapter kinesisInboundAdapter(AmazonKinesisAsync kinesisClient) {
        KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter(kinesisClient, streamName);
        adapter.setOutputChannel(kinesisInputChannel());
        adapter.setConsumerGroup(consumerGroup);
        adapter.setShardIteratorType(ShardIteratorType.TRIM_HORIZON); // Start reading from oldest records
        adapter.setBatchSize(10); // Adjust based on your needs
        // For single-instance testing, disable lease management to avoid DynamoDB dependencies
        adapter.setLeaseConsumer(null);
        return adapter;
    }

    @Bean
    public MessageChannel kinesisInputChannel() {
        return new DirectChannel();
    }

    // Service activator to handle incoming Kinesis messages
    @ServiceActivator(inputChannel = "kinesisInputChannel")
    public void processKinesisMessage(String payload) {
        System.out.println("Received Kinesis message: " + payload);
    }
}

Step 3: Diagnose Repeated Errors #

Here are the top causes for your retry loop:

• Missing Stream: Verify your Kinesis stream exists in the configured AWS region. A ResourceNotFoundException in the full log confirms this.
• Permission Issues: Your AWS credentials lack required permissions (kinesis:DescribeStream, kinesis:GetShardIterator, kinesis:GetRecords). Check your IAM policy for the attached user/role.
• Lease Management Conflicts: If running multiple instances without a DynamoDB lease manager, the consumer group will repeatedly rebalance. Use adapter.setLeaseConsumer(null) for single-instance testing, or configure a DynamoDB lease store for distributed deployments.
• Truncated Logs: The full error message (after the truncated a.i.k.KinesisMessageDri...) will tell you exactly what's failing — always check the complete stack trace.

End-to-End Working Example #

To get a fully functional setup:

1. Use the pom.xml and configuration above.
2. Create the Kinesis stream in your AWS account matching kinesis.stream.name.
3. Grant your AWS credentials the necessary Kinesis permissions via IAM.
4. Run your Spring Boot app — it should start consuming messages without repeated errors.

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