Início / coding / Perguntas / 79200676
Accepted
Shuffling2241
Asked: 2024-11-19 00:21:06 +0800 CST

Qual é o efeito de streamInitialSequence caso já existam pontos de verificação no DynamoDB?

  • 772

Gostaria de usar KinesisMessageDrivenChannelAdapterpara ler registros de um fluxo do Kinesis. Ao iniciar o aplicativo do consumidor pela primeira vez, gostaria que ele recebesse todos os registros que já existem no fluxo. Em inicializações subsequentes, no entanto, o aplicativo deve continuar lendo do último número de sequência de checkpoint vindo do DynamoDb.

Minha suposição está correta ao adapter.setStreamInitialSequence(KinesisShardOffset.trimHorizon())fornecer esse comportamento?

spring-integration

1 respostas

  1. Best Answer

    Isso funciona apenas para novos consumidores no grupo. Se já houver um checkpoint para esse grupo de consumidores e esse shard, então faremos assim:

            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);
            }
        }

    Então, ele vai consumir de um ponto de verificação armazenado.

    Se você quiser usar isso trimHorizon, então ligue para KinesisMessageDrivenChannelAdapter.resetCheckpoints().

    • 0

relate perguntas