[FLINK-30935][connector/kafka] Add kafka serializers version check when using SimpleVersionedSerializer

[chucheng92]

What is the purpose of the change

Add deserialize version check for kafka simple versioned serializers like other SimpleVersionedSerializer implementations in case of incompatible or corrupt state when restoring from checkpoint.

Brief change log

Add deserialize version check logic for kafka simple versioned serializers.

Verifying this change

Add cases in KafkaCommittableSerializerTest and KafkaWriterStateSerializerTest and KafkaPartitionSplitSerializerTest.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: yes
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[chucheng92]

@leonardBang @PatrickRen sorry to bother you, can someone of you have a look please?

[chucheng92]

@tzulitai hi, Gordon, can you help to take a look please?

[chucheng92]

@tzulitai hi, Gordon. It's been a long time, please let me know if you have any concerns, I will fix it ASAP. thanks.

[MartijnVisser]

@chucheng92 Can you please rebase your PR?

[chucheng92]

@chucheng92 Can you please rebase your PR?

Yes, I have rebased it and passed ci. If you have time, pls help to review it again. thanks a lot.

[MartijnVisser]

@mas-chen Do you want to do a review?