@Override
    public Object read(SourceRecord record) {
        Object obj = converter.read(record);
        if (obj == null) {
            return null;
        }
        Struct value = (Struct) record.value();
        Schema valueSchema = record.valueSchema();
        Schema beforeSchema = valueSchema.field(Envelope.FieldName.BEFORE).schema();
        Struct before = value.getStruct(Envelope.FieldName.BEFORE);

        Map<StringData, StringData> oldData = new HashMap<>();
        for (int i = 0; i < this.fieldNames.length; i++) {
            final String fieldName = this.fieldNames[i];
            final StringConverter stringConverter = this.stringConverters[i];
            final Field field = beforeSchema.field(fieldName);

            if (field == null) {
                oldData.put(StringData.fromString(fieldName), null);
            } else {
                final Object fieldValue = before.get(field);
                final Schema fieldSchema = field.schema();
                StringData strFieldValue = null;
                try {
                    final String str = stringConverter.convert(fieldValue, fieldSchema);
                    strFieldValue = StringData.fromString(str);
                } catch (Exception e) {
                    LOG.error(
                            "Failed to convert value "
                                    + fieldValue
                                    + " ("
                                    + fieldSchema.name()
                                    + ") to string.");
                }

                oldData.put(StringData.fromString(fieldName), strFieldValue);
            }
        }

        return new GenericArrayData(new Object[]{new GenericMapData(oldData)});
    }
Formatter for SQL string representation of a timestamp value (with UTC timezone).
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/table/OldFieldMetadataConverter.java