KSQL produce one null message after publishing the good one

Viewed 31

What I need to do with KSQL is, summarizing, send a message to a topic everytime I got a close event AFTER an open one for a file. I got fileName, status and 2 other fields that I just transport data. This is what I did so far:

CREATE STREAM SOURCESTREAM (
    "field1" varchar,
    "field2" varchar,
    "fileName" varchar,
"status" varchar
) WITH (kafka_topic='STATUS', value_format='AVRO');


CREATE TABLE SOURCETABLE WITH (value_format='AVRO') as select
    LATEST_BY_OFFSET("field1") as "field1",
    LATEST_BY_OFFSET("field2") as "field2",
    "fileName",
LATEST_BY_OFFSET("status",2) as "status"
    from SOURCESTREAM
    GROUP BY "fileName";


CREATE table FINALTABLE
AS SELECT "field1", "field2", "fileName" key
FROM SOURCETABLE WHERE "status"[1]='open' and "status"[2] = 'close';

CREATE STREAM TARGETSTREAM (
    "field1" varchar,
    "field2" varchar,
    "fileName" varchar KEY
) WITH (kafka_topic='FINALTABLE', value_format='AVRO');

CREATE STREAM FILETRANSFER
WITH (kafka_topic='ISCLOSE', value_format='AVRO')
AS SELECT "field1", "field2", "fileName" FROM TARGETSTREAM PARTITION BY NULL;

When I try this, it seems to work fine, until I realized that if I send 2 'close' messages, I get a complete null message. With no fields at all. If I send n 'close' messages nothing happens, only with the second one after an "open".

I also notice that if I query all of the streams and tables involved the second 'close' message only goes until "sourcetable", as expected.... but it produce a null message! What is happening here?!

0 Answers
Related