Guava Cache as ValueState in Flink

I am trying to de-duplicate events in my Flink pipeline. I am trying to do that using guava cache. My requirement is that, I want to de-duplicate over a 1 minute window. But at any given point I want to maintain not more than 10000 elements in the cache.

A small background on my experiment with Flink windowing:

• Tumbling Windows: I was able to implement this using Tumbling windows + custom trigger. But the problem is, if an element occurs in the 59th minute and 61st minute, it is not recognized as a duplicate.
• Sliding Windows: I also tried sliding window with 10 second overlap + custom trigger. But an element that came in the 55th second is part of 5 different windows and it is written to the sink 5 times.

Please let me know if I should not be seeing the above behavior with windowing.

Back to Guava:

I have Event which looks like this and a EventsWrapper for these events which looks like this. I will be getting a stream of EventsWrappers. I should remove duplicate Events across different EventsWrappers. Example if I have 2 EventsWrappers like below:

[EventsWrapper{id='ew1', org='org1', events=[Event{id='e1', name='event1'}, Event{id='e2', name='event2'}]}, EventsWrapper{id='ew2', org='org2', events=[Event{id='e1', name='event1'}, Event{id='e3', name='event3'}]}

I should emit as output the following:

[EventsWrapper{id='ew1', org='org1', events=[Event{id='e1', name='event1'}, Event{id='e2', name='event2'}]}, EventsWrapper{id='ew2', org='org2', events=[Event{id='e3', name='event3'}]}

i.e Making sure that e1 event is emitted only once assuming these two events are within the time and size requirements of the cache.

I created a RichFlatmap function where I initiate a guava cache and value state like this. And set the Guava cache in the value state like this. My overall pipeline looks like this.

But each time I try to update the guava cache inside the value state:

eventsState.value().put(eventId, true);

I get the following error:

java.lang.NullPointerException
    at com.google.common.cache.LocalCache.hash(LocalCache.java:1696)
    at com.google.common.cache.LocalCache.put(LocalCache.java:4180)
    at com.google.common.cache.LocalCache$LocalManualCache.put(LocalCache.java:4888)
    at events.piepline.DeduplicatingFlatmap.lambda$flatMap$0(DeduplicatingFlatmap.java:59)
    at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:176)

On further digging, I found out that the error is because the keyEquivalence inside the Guava cache is null. I checked by directly setting on the Guava cache(not through state, but directly on the cache) and that works fine.

I felt this could be because, ValueState is not able to serialize GuavaCache. So I added a Serializer like this and registered it like this:

env.registerTypeWithKryoSerializer((Class<Cache<String,Boolean>>)(Class<?>)Cache.class, CacheSerializer.class); But this din't help either.

I have the following questions:

1. Any idea what I might be doing wrong with the Guava cache in the above case.
2. Is what I am seeing with my Tumbling and Slinding windows implementation is what is expected or am I doing something wrong?
3. What will happen if I don't set the Guava Cache in ValueState, instead just use it as a plain object in the DeduplicatingFlatmap class and operate directly on the Guava Cache instead of operating through the ValueState? My understanding is, the Guava cache won't be part of the Checkpoint. So when the pipeline fails and restarts, the GuavaCahe would have lost all the values in it and it will be empty on restart. Is this understanding correct?

Thanks a lot in advance for the help.