@Override
    public void deserialize(Message<RowData> message, Collector<RowData> collector) throws IOException {
        // shortcut in case no output projection is required,
        // also not for a cartesian product with the keys
        if (keyDeserialization == null && !hasMetadata) {
            valueDeserialization.deserialize(message.getData(),
                    new MetricsCollector<>(collector, sourceMetricData));
            return;
        }
        BufferingCollector keyCollector = new BufferingCollector();

        // buffer key(s)
        if (keyDeserialization != null) {
            keyDeserialization.deserialize(message.getKeyBytes(), keyCollector);
        }

        // project output while emitting values
        outputCollector.inputMessage = message;
        outputCollector.physicalKeyRows = keyCollector.buffer;
        outputCollector.outputCollector = collector;
        if ((message.getData() == null || message.getData().length == 0) && upsertMode) {
            // collect tombstone messages in upsert mode by hand
            outputCollector.collect(null);
        } else {
            valueDeserialization.deserialize(message.getData(), new MetricsCollector<>(
                    outputCollector, sourceMetricData));
        }

        keyCollector.buffer.clear();
    }
A specific {@link PulsarDeserializationSchema} for {@link PulsarDynamicTableSource}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/DynamicPulsarDeserializationSchema.java