https://flink.apache.org/ logo
Join Slack
Powered by
I'm using event time. The time range is large enou...
# troubleshooting
d
I'm using event time. The time range is large enough to have had this triggered. I've also set it to 1 second but still didn't get the function called.
d
Event time timers depend on watermarks. How are the watermarks configured? Have you checked to see if they are advancing?
d
No I have not. What's the best way to check for that? Log messages or Flink UI?
Watermark is being set from a Kafka Source:
Copy code
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
Yeah doesn't look like my water mark is advancing. Here is the extractor:
Copy code
DataStreamSource<Transaction> kafkaStream =
                env.fromSource(source, WatermarkStrategy
                        .<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Transaction>() {
                            @Override
                            public long extractTimestamp(Transaction txn, long recordTimestamp) {
                                if (txn != null && txn.getDate() != null) {
                                    Date date = txn.getDate();
                                    // System.out.println("Extracting timestamp: " + timestamp);
                                    return date.toInstant().toEpochMilli();
                                }
                                // fallback to stream time
                                String pCode = txn.get(Transaction.PCODE);
                                System.out.printf("Unable to extract date from txn: %s. Defaulting to recordTimestamp\n", pCode);
                                return recordTimestamp;
                            }
                        }), "Kafka Source With Timestamps and Watermarks");
d
The most common cause of this problem is having idle or empty kafka partitions, or having the Flink parallelism set higher than the number of kafka partitions. In such cases the KafkaSource instances w/o any data to read won’t advance their watermarks, and this can hold back the watermarks globally. If this is the cause, you can fix it by using 
.withIdleness
or by reducing the parallelism.
d
Awesome that fixed it! Any theory on how to determine what Duration to set this too?
Do windows wait for the whole duration to complete before it emits a results or will it emit partial results? Just curious if what the default behavior is or if the idleness issue is why the window was not emitting.
d
By default windows do not emit partial results. Windows wait for a watermark large enough to signal that the window is complete. If you want to see partial results, you can use a custom trigger, or use Flink SQL with CUMULATE windows.
2 Views
Open in Slack
PreviousNext
Powered by