[
https://issues.apache.org/jira/browse/FLINK-30935?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17685136#comment-17685136
]
Ran Tao edited comment on FLINK-30935 at 2/9/23 7:49 AM:
---------------------------------------------------------
Hi [~chesnay] [~becket_qin] what do u think?
was (Author: lemonjing):
Hi [~chesnay] what do u think?
> 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
>
> {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.
> e.g.
> {code:java}
> @Override
> public KafkaPartitionSplit deserialize(int version, byte[] serialized) throws
> IOException {
> switch (version) {
> case 0:
> return deserializeV0(serialized);
> default:
> throw new IOException("Unrecognized version or corrupt state: " +
> version);
> }
> }
> private KafkaPartitionSplit deserializeV0(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}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
