KafkaConnect.OffsetStorageReader在第一次调用poll()时抛出异常，但在下一次调用poll()时可以获取到sourceOffset，不会出现异常

这个问题是由于在 OffsetStorageReader 创建时没有调用恰当的方法或配置所导致的。可以通过在创建 OffsetStorageReader 时添加以下代码解决问题：

Map<String, String> offsetConfig = new HashMap<>(); offsetConfig.put("bootstrap.servers", "localhost:9092"); offsetConfig.put("group.id", "test-group"); offsetConfig.put("key.deserializer", StringDeserializer.class.getName()); offsetConfig.put("value.deserializer", StringDeserializer.class.getName()); offsetConfig.put("auto.offset.reset", "earliest");

KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(offsetConfig); kafkaConsumer.subscribe(Arrays.asList("test-topic"));

// 为 OffsetStorageReader 添加以下代码 OffsetStorageReader offsetStorageReader = new KafkaOffsetStorageReader(kafkaConsumer);

// 确保在创建 OffsetStorageReader 之后调用 poll() 方法 try { while (true) { Map<String, Object> sourcePartition = new HashMap<>(); sourcePartition.put("topic", "test-topic"); Map<String, Object> sourceOffset = offsetStorageReader.offset(sourcePartition);

    // 在 OffsetStorageReader 调用 poll() 方法之前需要调用 setStartOffsets 方法
    kafkaConsumer.assignment().forEach(topicPartition -> {
        sourcePartition.put("partition", topicPartition.partition());
        offsetStorageReader.setStartOffsets(Collections.singletonMap(sourcePartition, topicPartition.offset()));
    });

    // 调用 poll() 方法
    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));

    // 处理数据记录...
}

} catch (Exception e) { e.printStackTrace(); } finally { kafkaConsumer.close(); }

使用 setStartOffsets() 方法和正确的参数为 OffsetStorageReader 指定正确的初始偏移量来解决这个问题。