Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/5393#discussion_r165586994 --- Diff: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java --- @@ -214,6 +226,7 @@ protected KinesisDataFetcher(List<String> streams, this.totalNumberOfConsumerSubtasks = runtimeContext.getNumberOfParallelSubtasks(); this.indexOfThisConsumerSubtask = runtimeContext.getIndexOfThisSubtask(); this.deserializationSchema = checkNotNull(deserializationSchema); + this.shardAssigner = checkNotNull(shardAssigner); --- End diff -- We also need to try cleaning the closure of the given object (if it is a non-static inner class): ``` ClosureCleaner.clean(shardAssigner, true); ```
---