Hello team, I am trying to implement a timestamp extractor that returns currentTimeMillis as the timestamp for each event (i.e., using ingestion time semantics).

watermarkStrategy = WatermarkStrategy.forMonotonousTimestamps()
        .withIdleness(Duration.ofSeconds(60))
        .withTimestampAssigner((event, ts) -> System.currentTimeMillis());

I am getting the following Exception: Caused by: java.lang.NoSuchMethodException: <className>.$deserializeLambda$(java.lang.invoke.SerializedLambda) at java.lang.Class.getDeclaredMethod(Class.java:2158) at java.lang.invoke.SerializedLambda$1.run(SerializedLambda.java:224) at java.lang.invoke.SerializedLambda$1.run(SerializedLambda.java:221) at java.security.AccessController.doPrivileged(Native Method) at java.lang.invoke.SerializedLambda.readResolve(SerializedLambda.java:221) at sun.reflect.GeneratedMethodAccessor18.invoke(Unknown Source) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1274) ... 30 more can somebody please point me to the correct way to resolve this?

I believe you need to get the type in there, e.g.,

WatermarkStrategy<Event> wmStrategy =
        WatermarkStrategy.<Event>forMonotonousTimestamps()
                .withIdleness(Duration.ofDays(7))
                .withTimestampAssigner((event, ts) -> System.currentTimeMillis());