I am running a Flink batch job on 1.16.1. The job is reading from GCS, doing a keyBy and keyedProcessFunction (for some business logic) and finally writing the output to the GCS sink. The job is able to read from GCS but failing in second stage (keyBy and process function) with an exception