CODEWITHSUNDEEP
apache-kafka-streams
glarus089
glarus089

Reputation: 93

Kafka Ktable querying

Trying to get records from "connect-configs" topic via KTable

    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);
        }
      }


};

ktable.queryableStoreName() always null. Why where is no store for querying? I see topics like "test_connect-configs_12-connect-configsSTATE-STORE-0000000000-changelog". How can I read records and how to get change events of the KTable state?

Upvotes: 3

Views: 5156

Answers (1)

Matthias J. Sax
Matthias J. Sax

Reputation: 62285

You need to specify a name for the underlying store when creating the KTable:

builder.table("connect-configs", Materialized.as<...>("my-store-name"));

Upvotes: 7

Related Questions