Можем ли мы выбрать определенную строку записей из слитной темы кафки?

Я предполагаю, что вы уже сделали

CREATE STREAM FOO_02 WITH (KAFKA_TOPIC='FOO_02', FORMAT='AVRO');

потому что иначе ваш SELECT потерпел бы неудачу.

Итак, поскольку PRINT успешно показывает, что в тема есть данные, вы можете запросить поток, используя нужный предикат. Единственное, что вам нужно сделать, это указать ksqlDB обрабатывать все данные в теме, а не только новые записи (что и делает from beginning в операторе PRINT). Для этого запустите:

SET 'auto.offset.reset' = 'earliest';

и потом запустить SELECT.

Редактировать

Can we select the only latest records? For example, I have multiple data being push to topic for COL1=0, but I only want to grab the latest one because its the newest data and it is the only correct one. something like where rowtime = max?

То, что вы описываете, это TABLE: последнее значение для данного ключа.

CREATE TABLE FOO AS
  SELECT COL1, 
         LATEST_BY_OFFSET(COL2) AS COL2
    FROM FOO_02
   WHERE COL1=0
   GROUP BY COL1;

В результирующей таблице будет одна запись для COL1 с последним значением COL2 по мере получения новых сообщений.

Any way to bring the old data in the Stream into the table as well?

Чтобы обработать существующие данные, установите смещение на самое раннее перед запуском оператора CREATE.

SET 'auto.offset.reset' = 'earliest';

CREATE TABLE FOO AS
[…]