@Override
    public void open(SerializationSchema.InitializationContext context) throws Exception {
        if (keySerialization != null) {
            keySerialization.open(context);
        }
        valueSerialization.open(context);
        if (partitioner != null) {
            partitioner.open(parallelInstanceId, numParallelInstances);
        }
        if (dirtySink != null) {
            dirtySink.open(new Configuration());
        }
        // Only support dynamic topic when the topicPattern is specified
        // and the valueSerialization is RawFormatSerializationSchema
        if (valueSerialization instanceof RawFormatSerializationSchema && StringUtils.isNotBlank(sinkMultipleFormat)) {
            multipleSink = true;
            jsonDynamicSchemaFormat =
                    (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(sinkMultipleFormat);
        }

        if (partitioner instanceof SingleTableCustomFieldsPartitioner) {
            ((SingleTableCustomFieldsPartitioner<?>) partitioner).setValueFieldGetters(valueFieldGetters);
        }
    }
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