public Set<SchemaChangeType> extractSchemaChangeType(JsonNode rootNode) {
        Operation operation;
        try {
            JsonNode operationNode = Preconditions.checkNotNull(rootNode.get("operation"),
                    "Operation node is null");
            operation = Preconditions.checkNotNull(
                    JsonDynamicSchemaFormat.OBJECT_MAPPER.convertValue(operationNode, new TypeReference<Operation>() {
                    }), "Operation is null");
        } catch (Exception e) {
            LOG.error("Extract Operation from origin data failed", e);
            return Collections.emptySet();
        }
        String originSchema = jsonDynamicSchemaFormat.extractDDL(rootNode);
        Set<SchemaChangeType> types = SchemaChangeUtils.extractSchemaChangeTypes(operation);
        if (types.isEmpty()) {
            LOG.warn("Unsupported for schema-change: {}", originSchema);
        }
        return types;
    }
Contains the position for each value of {@link WritableMetadata} in the
consumed row or -1 if this metadata key is not used.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/DynamicKafkaSerializationSchema.java