[ https://issues.apache.org/jira/browse/FLINK-30935?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Ran Tao updated FLINK-30935: ---------------------------- Description: {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} was: {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 in case of incompatible or corrupt state when restoring from checkpoint.. > 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)