@Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        final SerializationSchema<RowData> keySerialization =
                createSerialization(context, keyEncodingFormat, keyProjection, keyPrefix);

        final SerializationSchema<RowData> valueSerialization =
                createSerialization(context, valueEncodingFormat, valueProjection, null);

        final PulsarSerializationSchema<RowData> pulsarSerializer =
                createPulsarSerializer(keySerialization, valueSerialization);

        final SinkFunction<RowData> pulsarSink = createPulsarSink(
                this.topic,
                this.properties,
                pulsarSerializer);

        return SinkFunctionProvider.of(pulsarSink, parallelism);
    }
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