[jira] (FLINK-30935) Add KafkaSerializer deserialize check when using SimpleVersionedSerializer

["Ran Tao (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-30935 ]


    Ran Tao deleted comment on FLINK-30935:
    ---------------------------------

was (Author: lemonjing):
[~becket_qin] [~renqs] Hi, Can u help me to review the PR when you are free? thanks.

> Add KafkaSerializer deserialize check when using SimpleVersionedSerializer
> --------------------------------------------------------------------------
>
>                 Key: FLINK-30935
>                 URL: https://issues.apache.org/jira/browse/FLINK-30935
>             Project: Flink
>          Issue Type: Improvement
>          Components: Connectors / Kafka
>            Reporter: Ran Tao
>            Priority: Major
>              Labels: pull-request-available
>
> {code:java}
> @Override
> public int getVersion() {
>     return CURRENT_VERSION;
> }
> @Override
> public KafkaPartitionSplit deserialize(int version, byte[] serialized) throws IOException {
>     try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
>             DataInputStream in = new DataInputStream(bais)) {
>         String topic = in.readUTF();
>         int partition = in.readInt();
>         long offset = in.readLong();
>         long stoppingOffset = in.readLong();
>         return new KafkaPartitionSplit(
>                 new TopicPartition(topic, partition), offset, stoppingOffset);
>     }
> } {code}
> Current kafka many implemented serializers do not deal with version check. I think we can add it like many other connectors's implementation in case of incompatible or corrupt state when restoring from checkpoint.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)