@VisibleForTesting
    static TupleSerializer<Tuple2<KafkaTopicPartition, Long>> createStateSerializer(
            ExecutionConfig executionConfig) {
        // explicit serializer will keep the compatibility with GenericTypeInformation and allow to
        // disableGenericTypes for users
        TypeSerializer<?>[] fieldSerializers =
                new TypeSerializer<?>[]{
                        new KryoSerializer<>(KafkaTopicPartition.class, executionConfig),
                        LongSerializer.INSTANCE
                };
        @SuppressWarnings("unchecked")
        Class<Tuple2<KafkaTopicPartition, Long>> tupleClass =
                (Class<Tuple2<KafkaTopicPartition, Long>>) (Class<?>) Tuple2.class;
        return new TupleSerializer<>(tupleClass, fieldSerializers);
    }
Creates state serializer for kafka topic partition to offset tuple. Using of the explicit
state serializer with KryoSerializer is needed because otherwise users cannot use
'disableGenericTypes' properties with KafkaConsumer.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/FlinkKafkaConsumerBase.java