在 KafkaStreams 拓扑中重试消息

Question

我有一个 kafkaStreams 拓扑，其中有一个处理器 API。在处理器内部，有一个调用外部 API 的逻辑。

如果 API 返回 503，则需要重试消息。

现在，我正在尝试将此消息推送到不同的 kafka 主题并使用“Punctuate”方法每分钟从失败的主题中提取一批消息，重试。

有没有更好的方法/方法来解决这个问题？

Answer 1

另一种稳健的方法是使用状态存储。它们由 Kafka 支持为压缩的变更日志主题。

您可以将失败的消息存储在状态存储中，并通过调用 schedule (punctuate) 将它们全部处理，然后删除所有成功处理的消息。

例如：

public class MyProcessor {

    private final long schedulerIntervalMs = 60000;
    private final String entityStoreName = "failed-message-store";
    private KeyValueStore<String, Object> entityStore;

    @Override
    public void init(ProcessorContext context) {
        this.entityStore = (KeyValueStore) context().getStateStore(entityStoreName);
        context().schedule(Duration.ofMillis(this.schedulerIntervalMs), PunctuationType.WALL_CLOCK_TIME,
                timestamp -> processFailedMessagesStore());
    }

    @Override
    public void process(String key, Object value) {
        boolean apiCallSuccessful = // call API

        if (!apiCallSuccesfull) {
            entityStore.put(key, value);
        }
    }

    private void processFailedMessagesStore() {
        try (KeyValueIterator<String, Object> allItems = entityStore.all()) {
            allItems.forEachRemaining(item -> {
                boolean successfullyProcessed = // re-process
                
                if (successfullyProcessed) {
                    entityStore.delete(item.key);
                }
            });
        }
    }
}