private Object getValueWithSchema(Object fieldValue, String schemaName) {
        if (fieldValue == null || StringUtils.isBlank(schemaName)) {
            return fieldValue;
        }
        // Because Oracle CDC has been shaded, the schema will have the prefix
        // 'org.apache.inlong.sort.cdc.oracle.shaded' added,
        // so we need to use `schemaName.endsWith()` to determine the Schema type.
        if (schemaName.endsWith(MicroTime.SCHEMA_NAME)) {
            Instant instant = Instant.ofEpochMilli((Long) fieldValue / 1000);
            fieldValue = timeFormatter.format(LocalDateTime.ofInstant(instant, ZONE_UTC));
        } else if (schemaName.endsWith(Date.SCHEMA_NAME)) {
            fieldValue = dateFormatter.format(LocalDate.ofEpochDay((Integer) fieldValue));
        } else if (schemaName.endsWith(ZonedTimestamp.SCHEMA_NAME)) {
            ZonedDateTime zonedDateTime = ZonedDateTime.parse((CharSequence) fieldValue);
            fieldValue = zonedDateTime.withZoneSameInstant(serverTimeZone).toLocalDateTime()
                    .atZone(ZONE_UTC).format(DateTimeFormatter.ISO_INSTANT);
        } else if (schemaName.endsWith(Timestamp.SCHEMA_NAME)) {
            Instant instantTime = Instant.ofEpochMilli((Long) fieldValue);
            fieldValue = LocalDateTime.ofInstant(instantTime, ZONE_UTC).toString();
        } else if (schemaName.endsWith(MicroTimestamp.SCHEMA_NAME)) {
            Instant instantTime = Instant.ofEpochMilli((Long) fieldValue / 1000);
            fieldValue = LocalDateTime.ofInstant(instantTime, ZONE_UTC).toString();
        }
        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/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/debezium/table/RowDataDebeziumDeserializeSchema.java