yuxiqian commented on code in PR #4374:
URL: https://github.com/apache/flink-cdc/pull/4374#discussion_r3063487933
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/data/RecordDataSerializer.java:
##########
@@ -33,43 +34,86 @@ public class RecordDataSerializer extends
TypeSerializerSingleton<RecordData> {
private static final long serialVersionUID = 1L;
+ /** Type tag for BinaryRecordData serialization. */
+ private static final byte BINARY_RECORD_TYPE = 0;
+
+ /** Type tag for GenericRecordData serialization. */
+ private static final byte GENERIC_RECORD_TYPE = 1;
+
private final BinaryRecordDataSerializer binarySerializer =
BinaryRecordDataSerializer.INSTANCE;
public static final RecordDataSerializer INSTANCE = new
RecordDataSerializer();
@Override
public RecordData createInstance() {
- // BinaryRecordData is the only implementation of RecordData
return new BinaryRecordData(1);
}
@Override
public void serialize(RecordData recordData, DataOutputView target) throws
IOException {
- // BinaryRecordData is the only implementation of RecordData
- binarySerializer.serialize((BinaryRecordData) recordData, target);
+ if (recordData instanceof BinaryRecordData) {
+ target.writeByte(BINARY_RECORD_TYPE);
+ binarySerializer.serialize((BinaryRecordData) recordData, target);
+ } else if (recordData instanceof GenericRecordData) {
+ target.writeByte(GENERIC_RECORD_TYPE);
+ GenericRecordDataSerializer.serialize((GenericRecordData)
recordData, target);
+ } else {
+ throw new IOException(
+ "Unsupported RecordData type: " +
recordData.getClass().getName());
+ }
}
@Override
public RecordData deserialize(DataInputView source) throws IOException {
- // BinaryRecordData is the only implementation of RecordData
- return binarySerializer.deserialize(source);
+ byte type = source.readByte();
+ if (type == BINARY_RECORD_TYPE) {
+ return binarySerializer.deserialize(source);
+ } else if (type == GENERIC_RECORD_TYPE) {
+ return GenericRecordDataSerializer.deserialize(source);
+ } else {
+ throw new IOException("Unknown RecordData type tag: " + type);
+ }
Review Comment:
Should be fine, as `DataChangeEvent`s are not persisted in states. They're
just [de]serialized for network transferring.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
