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?!