private PulsarSerializationSchema<RowData> createPulsarSerializer(SerializationSchema<RowData> keySerialization,
            SerializationSchema<RowData> valueSerialization) {
        final List<LogicalType> physicalChildren = physicalDataType.getLogicalType().getChildren();

        final RowData.FieldGetter[] keyFieldGetters = Arrays.stream(keyProjection)
                .mapToObj(targetField -> RowData.createFieldGetter(physicalChildren.get(targetField), targetField))
                .toArray(RowData.FieldGetter[]::new);

        final RowData.FieldGetter[] valueFieldGetters = Arrays.stream(valueProjection)
                .mapToObj(targetField -> RowData.createFieldGetter(physicalChildren.get(targetField), targetField))
                .toArray(RowData.FieldGetter[]::new);

        // determine the positions of metadata in the consumed row
        final int[] metadataPositions = Stream.of(WritableMetadata.values())
                .mapToInt(m -> {
                    final int pos = metadataKeys.indexOf(m.key);
                    if (pos < 0) {
                        return -1;
                    }
                    return physicalChildren.size() + pos;
                })
                .toArray();

        // check if metadata is used at all
        final boolean hasMetadata = metadataKeys.size() > 0;

        final long delayMilliseconds = Optional.ofNullable(this.properties
                .getProperty(PulsarOptions.SEND_DELAY_MILLISECONDS, "0"))
                .filter(StringUtils::isNumeric)
                .map(Long::valueOf)
                .orElse(0L);

        return new DynamicPulsarSerializationSchema(
                keySerialization,
                valueSerialization,
                keyFieldGetters,
                valueFieldGetters,
                hasMetadata,
                metadataPositions,
                upsertMode,
                DataTypeUtils.projectRow(physicalDataType, valueProjection),
                formatType,
                delayMilliseconds);
    }
Metadata that is appended at the end of a physical sink row.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarDynamicTableSink.java