KTable data not being printed

Kafka Streams
1

I have a Ktable with I create after consuming a topic. I am unable to print the values of
this table.

Have attached screenshots

image
image1933×1080 177 KB

2

The screenshot is very hard to read… However, “printing KTable” data sound a little “odd”. In the end, using Kafka Streams you assembly a dataflow program that you deploy…

Thus, you could for example print a message each time the table is updated:

myTable.toStream().print();

Or you can use “Interactive Queries” to tap into the table state from “outside”. Cf. Kafka Streams Interactive Queries | Confluent Documentation

3

Thanks for responding.

I have a few questions:

  1. I am creating a K-Table out of a topic which is in JSON. So do I need to specifically write a logic where I write entries to the K-Table as we do in SQL?
KTable<String, String> kTable = builder.table("testqa2", Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("queryableKTable")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()));

This is the way I am creating my K-Table, where testqa2 is my topic

To print a stream out of this I:

        KStream<String, String> output = kTable.toStream();
        output.print(Printed.toSysOut());

When I run this code, the sysOut isn’t triggered. The debug has Ktable has all-byte values.

Please can you help me what to do print the get result of the output.print(…) part in String/JSON?

Thanks,
Anmol

4

While trying the print function, it seems it’s deprecated.

kTable.toStream().print();

image

Can we still make use of it?

5

Not sure what version you are using, but the JavaDocs should always point to the new method that should be used if anything is deprecated. For this case, there should be print(Printed) overload.

In general, you may use deprecated methods, but they might get removed in the next major release and thus, you should updated your code when possible to allow yourself to upgrade to newer versions without friction.

6

Thanks for responding.

I have a few questions:

  1. I am creating a K-Table out of a topic which is in JSON. So do I need to specifically write a logic where I write entries to the K-Table as we do in SQL?
KTable<String, String> kTable = builder.table("testqa2", Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("queryableKTable")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()));

This is the way I am creating my K-Table, where testqa2 is my topic

To print a stream out of this I:

        KStream<String, String> output = kTable.toStream();
        output.print(Printed.toSysOut());

When I run this code, the sysOut isn’t triggered. The debug has Ktable has all-byte values.

Please can you help me what to do print the get result of the output.print(…) part in String/JSON?

Thanks,
Anmol

Would you be having any idea about this?

7

As pointed out already, the question is, if the data is in the key or in the value of the record. Your statement does not say anything about it…

If you record key is null, and your JSON data is only in the value, you cannot read the topic as a table, because a table requires a key. To set a key, you could first read the topic as a stream, set a key, and convert it into a table:

KTable tabe = builder.stream(...).selectKey(...).toTable(...);

So do I need to specifically write a logic where I write entries to the K-Table as we do in SQL?

Don’t understand this question.

For the code snippet you showed: if the records don’t have a key, they would be dropped on read, and thus no output is printed… (Btw: because of record caching, even if there is output, it may take 30 seconds by default until you see any output. Cf. Kafka Streams Memory Management | Confluent Documentation)