Hi Team, I'm missing out on some fundamental aspect. Just adding a

filter

on kinesis stream and then

sink

to JDBC database

kinesisStream.filter(event -> EventType.AGENT_START == event.getType())
        .addSink(insertAssetSink);

No data gets saved to DB. If I remove the

filter

operation, data gets saved. Has it got something to do with whether I'm processing all data from the kinesis or not ?

since removing filter pushes record to sink, it is unlikely the issue is with source IMO. have you verified when removing filter that some data passes the filter predicate? also I assume EventType is an enum?

Yes. If I do

.print()

after the

.filter()

, it prints out the records I expect to see. But if I change from

.print()

to

.addSink

, nothing happens with those records