private Object getValueWithSchema(Object fieldValue, String schemaName) {
        if (fieldValue == null) {
            return null;
        }
        switch (schemaName) {
            case MicroTime.SCHEMA_NAME:
                Instant instant = Instant.ofEpochMilli((Long) fieldValue / 1000);
                fieldValue = timeFormatter.format(LocalDateTime.ofInstant(instant, ZONE_UTC));
                break;
            case Date.SCHEMA_NAME:
                fieldValue = dateFormatter.format(LocalDate.ofEpochDay((Integer) fieldValue));
                break;
            case ZonedTimestamp.SCHEMA_NAME:
                ZonedDateTime zonedDateTime = ZonedDateTime.parse((CharSequence) fieldValue);
                fieldValue = zonedDateTime.withZoneSameInstant(serverTimeZone).toLocalDateTime()
                        .atZone(ZONE_UTC).format(DateTimeFormatter.ISO_INSTANT);
                break;
            case Timestamp.SCHEMA_NAME:
                Instant instantTime = Instant.ofEpochMilli((Long) fieldValue);
                fieldValue = LocalDateTime.ofInstant(instantTime, ZONE_UTC).toString();
                break;
            case MicroTimestamp.SCHEMA_NAME:
                instantTime = Instant.ofEpochMilli((Long) fieldValue / 1000);
                fieldValue = LocalDateTime.ofInstant(instantTime, ZONE_UTC).toString();
                break;
            case Decimal.LOGICAL_NAME:
                // no need to transfer decimal type since the value is already decimal
                break;
            default:
                LOG.debug("schema {} is not being supported", schemaName);
        }
        return fieldValue;
    }
extract the data with the format provided by debezium

@param fieldValue
@param schemaName
@return the extracted data with schema
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java