AWS运动-如何从最后一个检查点恢复消费

Question

我正在使用KCL（v2）将Kafka消费者转换为AWS动觉消费者。在Kafka中，偏移量用于帮助消费者跟踪其最近使用的消息。如果我的Kafka应用程序死机，它将使用重新启动时停止的偏移量。

然而，这在Kinesis中是不一样的。我可以设置动力学ClientLibConfiguration.withfiralPositionInStream（...），但唯一的参数是TRIM_HORIZON、LATEST或AT_TIMESTAMP。如果我的Kinesis应用程序死机，它将不知道重新启动时从哪里恢复消费。

我的KCL消费者非常简单。方法如下所示：

KinesisClientLibConfiguration config = new KinesisClientLibConfiguration("benTestApp",
            "testStream", new DefaultAWSCredentialsProviderChain(), UUID.randomUUID().toString());
config.withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);

Worker worker = new Worker.Builder()
            .recordProcessorFactory(new KCLRecordProcessorFactory())
            .config(config)
            .build();

而RecordProcessor是一个简单的实现：

@Override
public void initialize(InitializationInput initializationInput) {
    LOGGER.info("Initializing record processor for shard: {}", initializationInput.getShardId());
}

@Override
public void processRecords(ProcessRecordsInput processRecordsInput) {
    List<Record> records = processRecordsInput.getRecords();
    LOGGER.info("Retrieved {} records", records.size());
    records.forEach(r -> LOGGER.info("Record: {}", StandardCharsets.UTF_8.decode(r.getData())));
}

@Override
public void shutdown(ShutdownInput shutdownInput) {
    LOGGER.info("Shutting down input");
}

如果我检查相应的DynamoDB表，则检查点的值设置为TRIM\u HORIZON，并且不会随着记录的使用而使用SequenceID进行更新。

这里有什么解决方案来确保我消费每条消息？

Answer 1

正如@kdgregory所指出的，KCL要求用户设置自己的检查点。工作代码：

@Override
public void initialize(InitializationInput initializationInput) {
    LOGGER.info("Initializing record processor for shard: {}", initializationInput.getShardId());
}

@Override
public void processRecords(ProcessRecordsInput processRecordsInput) {
    List<Record> records = processRecordsInput.getRecords();
    LOGGER.info("Retrieved {} records", records.size());
    records.forEach(r -> LOGGER.info("Record with sequenceId {} at date {} : {}", r.getSequenceNumber(),
            r.getApproximateArrivalTimestamp(), StandardCharsets.UTF_8.decode(r.getData())));
    try {
        processRecordsInput.getCheckpointer().checkpoint();
    } catch (InvalidStateException | ShutdownException e) {
        LOGGER.error("Unable to checkpoint");
    }
}

@Override
public void shutdown(ShutdownInput shutdownInput) {
    LOGGER.info("Shutting down input");
    try {
        shutdownInput.getCheckpointer().checkpoint();
    } catch (InvalidStateException | ShutdownException e) {
        LOGGER.error("Unable to checkpoint");
    }
}