java - 使用 Spring Cloud Stream 寻求 Kafka 偏移量

Question

我有一个事件源服务，用于监听 Kafka 主题并将状态保存在关系数据库中。

考虑该服务的合适恢复策略(即如何在灾难恢复场景中恢复数据库)，一种选择是保存数据库中的当前偏移量，拍摄快照，然后从快照恢复。在这种情况下，服务在“恢复模式”下启动时需要寻找偏移量。

我正在使用 Spring Cloud Stream，想知道该框架是否提供了任何寻找偏移量的机制？

我意识到恢复的另一种选择是简单地从头开始播放所有事件，但这对于我的某些微服务来说并不是理想的选择。

Answer 1

有 KafkaBindingRebalanceListener 接口(interface)可供使用

@Slf4j
@Component
public class KafkaRebalanceListener implements KafkaBindingRebalanceListener {

    @Value("${config.kafka.topics.offsets:null}")
    private String topicOffsets;

    @Override
    public void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer, Collection<TopicPartition> partitions, boolean initial) {
        if (topicOffsets != null && initial) {
            final Optional<Map<TopicPartition, Long>> offsetsOptional = parseOffset(topicOffsets);
            if (offsetsOptional.isPresent()) {
                final Map<TopicPartition, Long> offsetsMap = offsetsOptional.get();
                partitions.forEach(tp -> {
                    if (offsetsMap.containsKey(tp)) {
                        final Long offset = offsetsMap.get(tp);
                        try {
                            log.info("Seek topic {} partition {} to offset {}", tp.topic(), tp.partition(), offset);
                            consumer.seek(tp, offset);
                        } catch (Exception e) {
                            log.error("Unable to set offset {} for topic {} and partition {}", offset, tp.topic(), tp.partition());
                        }
                    }
                });
            }
        }
    }

    private Optional<Map<TopicPartition, Long>> parseOffset(String offsetParam) {
        if (offsetParam == null || offsetParam.isEmpty()) {
            return Optional.empty();
        }

        return Optional.of(Arrays.stream(offsetParam.split(","))
                .flatMap(slice -> {
                    String[] items = slice.split("\\|");
                    String topic = items[0];
                    return Arrays.stream(Arrays.copyOfRange(items, 1, items.length))
                            .map(r -> {
                                String[] record = r.split(":");
                                int partition = Integer.parseInt(record[0]);
                                long offset = Long.parseLong(record[1]);
                                return new AbstractMap.SimpleEntry<>(new TopicPartition(topic, partition), offset);
                            });
                }).collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue)));
    }
}

config.kafka.topics.offsets 字段如下所示，但您可以使用任何格式

String topicOffsets = "topic2|1:100|2:120|3:140,topic3|1:1000|2:1200|3:1400";