I would like to use KinesisMessageDrivenChannelAdapter to read records from a Kinesis stream. When starting the consumer application for the very first time, I would like it to receive all records that already exist in the stream. On subsequent starts though, the application should continue reading from the latest checkpointed sequence number coming from DynamoDb.

Is my assumption correct that adapter.setStreamInitialSequence(KinesisShardOffset.trimHorizon()) provides this behaviour?

I would like to use KinesisMessageDrivenChannelAdapter to read records from a Kinesis stream. When starting the consumer application for the very first time, I would like it to receive all records that already exist in the stream. On subsequent starts though, the application should continue reading from the latest checkpointed sequence number coming from DynamoDb.

Is my assumption correct that adapter.setStreamInitialSequence(KinesisShardOffset.trimHorizon()) provides this behaviour?

• spring-integration

1 Answer 1

That works only for new consumers in the group. If there is already a checkpoint for this consumer group and that shard, then we go like this:

        if (this.shardOffset.isReset()) {
            this.checkpointer.remove();
        }
        else {
            String checkpoint = this.checkpointer.getCheckpoint();
            if (checkpoint != null) {
                this.shardOffset.setSequenceNumber(checkpoint);
                this.shardOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
            }
        }

So, it is going to consume from a stored checkpoint.

If you'd like to use that trimHorizon, then you call KinesisMessageDrivenChannelAdapter.resetCheckpoints().