The problem:

I have three instances of a java application running in Kubernetes. My application uses Apache Camel to read from a Kinesis stream. I'm currently observing two related issues:

1. Each of the three running instances of my application is processing the records coming into the stream, when I only want each record to be processed once (I want three up and running for scaling purposes). I was hoping that while one instance is processing record A, a second could be picking up record B, etc.

2. Every time my application is re-deployed in Kubernetes, each instance starts every record all over again (in other words, it has no idea where it left off or which records have previously been processed).

3. After 5 minutes, the shard iterator that my application is using to poll kinesis times out. I know that this is normal behavior, but what I don't understand is why my application is not grabbing a new iterator. This screenshot shows the error from DataDog.

What I've tried: First off, I believe that this issue is caused by inconsistent shard iterator ids, and kinesis consumer ids across the three instances of my application, and across deploys. However, I have been unable to locate where these values are set in code, and how I could go about setting them. Of course, there may also be a better solution altogether. I have found very little documentation on Kinesis/Kubernetes/Camel working together, and so very little outside sources have been helpful.

The documentation on AWS Kinesis :: Apache Camel is very limited, but what I have tried playing around with the iterator type and building a custom Client Configuration.

Let me know if you need any additional information, thanks.

Configuring the client:

main.bind("kinesisClient", AmazonKinesisClientBuilder.defaultClient());
        .
        .
        .
    
inputUri = String.format("aws-kinesis://%s?amazonKinesisClient=#kinesisClient", rawKinesisName);
    
main.configure().addRoutesBuilder(new RawDataRoute(inputUri, inputTransform));

My route:

public class RawDataRoute extends RouteBuilder {
    private static final Logger LOG = new Logger(RawDataRoute.class, true);
    private String rawDataStreamUri;
    private Expression transform;

    public RawDataRoute(final String rawDataStreamUri, final Expression transform) {
        this.rawDataStreamUri = rawDataStreamUri;
        this.transform = transform;
    }

    @Override
    public void configure() {
        // TODO add error handling
        from(rawDataStreamUri)
            .routeId("raw_data_stream")
            .transform(transform)
            .to("direct:main_input_stream");
    }

}