private DeserializationRuntimeConverter getMultipleMigrationConverter(
            ZoneId serverTimeZone, DeserializationRuntimeConverterFactory userDefinedConverterFactory) {
        return new DeserializationRuntimeConverter() {

            private static final long serialVersionUID = 1L;

            @Override
            public Object convert(Object dbzObj, Schema schema) throws Exception {
                ConnectSchema connectSchema = (ConnectSchema) schema;
                List<Field> fields = connectSchema.fields();

                Map<String, Object> data = new HashMap<>();
                Struct struct = (Struct) dbzObj;

                for (Field field : fields) {
                    String fieldName = field.name();
                    Object fieldValue = struct.getWithoutDefault(fieldName);
                    Schema fieldSchema = schema.field(fieldName).schema();
                    String schemaName = fieldSchema.name();
                    if (schemaName != null) {
                        // normal type doesn't have schema name
                        // schema names are time schemas
                        fieldValue = getValueWithSchema(fieldValue, schemaName);
                    }
                    data.put(fieldName, fieldValue);
                }

                GenericRowData row = new GenericRowData(1);
                row.setField(0, data);

                return row;
            }

            @Override
            public Object convert(Object dbzObj, Schema schema, TableChange tableSchema) throws Exception {
                ConnectSchema connectSchema = (ConnectSchema) schema;
                List<Field> fields = connectSchema.fields();

                Map<String, Object> data = new HashMap<>();
                Struct struct = (Struct) dbzObj;

                for (Field field : fields) {
                    String fieldName = field.name();
                    Object fieldValue = struct.getWithoutDefault(fieldName);
                    Schema fieldSchema = schema.field(fieldName).schema();
                    String schemaName = fieldSchema.name();

                    // struct type convert normal type
                    if (fieldValue instanceof Struct) {
                        Column column = tableSchema.getTable().columnWithName(fieldName);
                        LogicalType logicType = RecordUtils.convertLogicType(column, (Struct) fieldValue);
                        DeserializationRuntimeConverter fieldConverter = createConverter(
                                logicType,
                                serverTimeZone,
                                userDefinedConverterFactory);
                        fieldValue =
                                convertField(fieldConverter, fieldValue, fieldSchema);
                        if (fieldValue instanceof DecimalData) {
                            fieldValue = ((DecimalData) fieldValue).toBigDecimal();
                        }
                        if (fieldValue instanceof TimestampData) {
                            fieldValue = ((TimestampData) fieldValue).toTimestamp();
                        }
                    }
                    if (schemaName != null) {
                        fieldValue = getValueWithSchema(fieldValue, schemaName);
                    }
                    if (fieldValue instanceof ByteBuffer) {
                        fieldValue = new String(((ByteBuffer) fieldValue).array());
                    }

                    data.put(fieldName, fieldValue);
                }

                GenericRowData row = new GenericRowData(1);
                row.setField(0, data);

                return row;
            }
        };
    }
Creates a runtime converter which assuming input object is not null.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java