TimerException in Flink Process

Question

We are running a Flink job using several operators including map, windowing, flatMap() and the job fails with the following error - just wondering what causes this error:

2021-05-27 07:00:07,023 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Collect Inventory (1/1)#0 (34d81cf2e59f350886f93a1e0f734d38) switched from RUNNING to FAILED with failure cause: org.apache.flink.streaming.runtime.tasks.AsynchronousException: Caught exception while processing timer.
 at org.apache.flink.streaming.runtime.tasks.StreamTask$StreamTaskAsyncExceptionHandler.handleAsyncException(StreamTask.java:1282)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.handleAsyncException(StreamTask.java:1258)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1397)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$null$16(StreamTask.java:1386)
 at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
 at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
 at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:344)
 at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:330)
 at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:202)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:661)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623)
 at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:776)
 at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
 at java.lang.Thread.run(Thread.java:748)
Caused by: TimerException{java.lang.RuntimeException: Assigned key must not be null!}
 ... 12 more
Caused by: java.lang.RuntimeException: Assigned key must not be null!
 at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:109)
 at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:93)
 at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:44)
 at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50)
 at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28)
 at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:50)
 at org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunction.apply(PassThroughWindowFunction.java:35)
 at org.apache.flink.streaming.runtime.operators.windowing.functions.InternalSingleValueWindowFunction.process(InternalSingleValueWindowFunction.java:48)
 at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.emitWindowContents(WindowOperator.java:577)
 at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.onProcessingTime(WindowOperator.java:533)
 at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:284)
 at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1395)
 ... 11 more
Caused by: java.lang.NullPointerException: Assigned key must not be null!
 at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:76)
 at org.apache.flink.runtime.state.KeyGroupRangeAssignment.assignKeyToParallelOperator(KeyGroupRangeAssignment.java:51)
 at org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.selectChannel(KeyGroupStreamPartitioner.java:63)
 at org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.selectChannel(KeyGroupStreamPartitioner.java:35)
 at org.apache.flink.runtime.io.network.api.writer.ChannelSelectorRecordWriter.emit(ChannelSelectorRecordWriter.java:54)
 at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:107)
 ... 22 more
2021-05-27 07:00:07,023 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Triggering cancellation of task code Collect Inventory (1/1)#0 (34d81cf2e59f350886f93a1e0f734d38).
{eventTime:2021-05-27T07:00:05.878Z, batchId:4a0c09ad-1e28-4f74-a19a-8e1422c5bc5a, serialDetail:null, inventoryCount:{"location": "tmobile01", "sku": "190198496225", "state": "LOST", "inventoryStatus": "Adjusted-Out", "inventoryType": "ACC", "sellableFlag": false, "version": 1, "updateTimestamp": 2021-05-27T07:00:07.118Z, "globalQuantity": -1, "localQuantity": -1, "eventId": null, "movementSource": "", "locationType": "Store"

Answer 1

It seems that You are using keyBy operation and the extracted key is null. You can't really have null key in Flink.