在自动递增数据库中的主键字段作为递增字段的情况下,kafka-connect-jdbc 在丢失和重复行方面是否安全?
尝试通过KTable从“ connect-configs”主题获取记录
public static void main(String... args) throws InterruptedException {
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test_connect-configs_12");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "***:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.Bytes().getClass());
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
StreamsBuilder builder = new StreamsBuilder();
KTable<String, Object> ktable = builder.table("connect-configs");
KafkaStreams streams = new KafkaStreams(builder.build(),config);
streams.cleanUp();
streams.start();
System.out.println(ktable.queryableStoreName());
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
ReadOnlyKeyValueStore<String, Object> view;
while (true) {
try {
System.out.println(ktable.queryableStoreName());
view = streams.store(ktable.queryableStoreName(), QueryableStoreTypes.keyValueStore());
} catch (InvalidStateStoreException ignored) {
// store not yet ready for querying
Thread.sleep(100);
}
}
};
Run Code Online (Sandbox Code Playgroud)
ktable.queryableStoreName()始终为null。为什么在哪里没有商店可以查询?我看到类似“ test_connect-configs_12-connect-configsSTATE-STORE-0000000000-changelog”的主题。如何读取记录以及如何获取KTable状态的更改事件?