AWS Kinesis Client Java：在 Stream 中设置 TRIM_HORIZON 位置不起作用

Question

我正在运行一个测试系统，该系统生成一个 Kinesis 生产者，该生产者开始写入消息，例如：1 到 100 到具有两个分片的流。

在该循环期间，消费者开始从流中读取消息。我注意到消费者只读取运行后进入流的LATEST 消息。例如，它从第 43 条消息开始读取。我尝试修改 Worker.class 以使用 TRIM_HORIZON 策略，但这似乎不起作用。

KinesisClientLibConfiguration c = new KinesisClientLibConfiguration("MediaPlan", "randeepstream",
    DefaultAWSCredentialsProviderChain.getInstance(),
    "consumer1")
    .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);
final Worker w = new Worker.Builder()
    .recordProcessorFactory(rpf)
    .config(kinesisConfig)
    .build();
new Thread(() -> w.run()).start();

我的消费者的处理器设置为：

public class ConsumerRecordProcessorImpl implements IRecordProcessor {

    public void initialize(InitializationInput initializationInput) {
        log.info("Setting up consumer with shard {} starting at {}", initializationInput.getShardId(),
                initializationInput.getExtendedSequenceNumber());
    }

    public void processRecords(ProcessRecordsInput processRecordsInput) {
        ...
    }
}

我希望看到如下消息： Setting up consumer with shard shardId-000000000000 starting at TRIM_HORIZON 0 但相反，我得到： Setting up consumer with shard shardId-000000000000 starting at LATEST 0

如何让我的消费者停止阅读最新消息并阅读所有未处理的消息？

Answer 1

这是一个使用 amazon-kinesis-client lib v2 的示例。

您将不得不使用 Schedular(software.amazon.kinesis.coordinator) 在后台读取记录并向此调度程序提供检索配置，如下所示

RetrievalConfig retrievalConfig = setRetrievalConfig();

Scheduler scheduler = new Scheduler(
        configsBuilder.checkpointConfig(),
        configsBuilder.coordinatorConfig(),
        configsBuilder.leaseManagementConfig(),
        configsBuilder.lifecycleConfig(),
        configsBuilder.metricsConfig(),
        configsBuilder.processorConfig(),
        retrievalConfig);

private RetrievalConfig setRetrievalConfig(){
    InitialPositionInStreamExtended initialPositionInStreamExtended = InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON);
    RetrievalConfig retrievalConfig = configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient));
    retrievalConfig.initialPositionInStreamExtended(initialPositionInStreamExtended);
    return retrievalConfig;
}

注意InitialPositionInStream.TRIM_HORIZON 这将告诉调度程序在最后一个已知位置之后开始使用记录。所以即使消费者宕机，生产者还在运行，消费者宕机期间产生的所有记录都会被消费掉。

注意：configBuilder 是 ConfigsBuilder (software.amazon.kinesis.common) 的对象

更新：initialPositionInStream 位置不会更新，除非您在处理从 kinesis 收到的数据后调用 checkpoint() API。 因此，一旦您调用 checkpinter()，流处理记录的最新位置就会在 DynamoDB 中更新，现在 KCL 将从该位置处理记录。