自定义日志包装器无法作用于KafkaSource顶点的技术问询

我来帮你梳理下问题根源，再给出几个针对性的解决方案：

为什么普通顶点生效但KafkaSource不行？ #

普通应用顶点的处理器是你自己实现的Processor，PeekWrappedP这类包装器用装饰器模式直接包裹你的Processor实例，拦截processElement方法就能统计单条消息的处理耗时。但KafkaSource作为源顶点，它的核心处理逻辑是Flink框架内部实现的（比如KafkaSourceReader拉取消息、反序列化等环节），并没有暴露给你可直接用普通Processor包装器拦截的入口，所以你的自定义包装器没法渗透进去。

解决方案1：自定义反序列化器统计单条消息处理耗时 #

如果你的核心需求是记录Kafka消息从字节数组反序列化为业务对象的耗时（这是KafkaSource处理消息的关键环节），可以通过包装DeserializationSchema实现：

public class TimingDeserializationSchema<T> implements DeserializationSchema<T> {
    private final DeserializationSchema<T> delegate;
    private static final Logger LOG = LoggerFactory.getLogger(TimingDeserializationSchema.class);

    public TimingDeserializationSchema(DeserializationSchema<T> delegate) {
        this.delegate = delegate;
    }

    @Override
    public T deserialize(byte[] message) throws IOException {
        long startTime = System.currentTimeMillis();
        T result = delegate.deserialize(message);
        long duration = System.currentTimeMillis() - startTime;
        // 可根据需求补充消息offset、key等标识，方便定位
        LOG.info("KafkaSource 反序列化单条消息耗时 {} ms，消息长度 {}", duration, message.length);
        return result;
    }

    @Override
    public boolean isEndOfStream(T nextElement) {
        return delegate.isEndOfStream(nextElement);
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return delegate.getProducedType();
    }
}

使用时直接替换原有的反序列化器：

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("your-topic")
        .setGroupId("your-group-id")
        // 用自定义计时包装器包裹原反序列化器
        .setValueOnlyDeserializer(new TimingDeserializationSchema<>(new SimpleStringSchema()))
        .build();

解决方案2：包装SourceReader统计批量拉取+处理耗时 #

如果需要统计KafkaSource从拉取一批消息到交付给下游的整体耗时，可以通过KafkaSource提供的decorateReader方法，包装它内部的SourceReader：

// 创建KafkaSource时注入Reader包装器
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("your-topic")
        .setGroupId("your-group-id")
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .decorateReader((ctx, config, delegateReader) -> new TimingSourceReader<>(delegateReader))
        .build();

// 自定义的SourceReader包装类
public class TimingSourceReader<T> implements SourceReader<T, KafkaRecordPartitionState> {
    private final SourceReader<T, KafkaRecordPartitionState> delegate;
    private static final Logger LOG = LoggerFactory.getLogger(TimingSourceReader.class);

    public TimingSourceReader(SourceReader<T, KafkaRecordPartitionState> delegate) {
        this.delegate = delegate;
    }

    @Override
    public void pollNext(ReaderOutput<T> output) throws Exception {
        long startTime = System.currentTimeMillis();
        // 委托原始Reader执行拉取和处理逻辑
        delegate.pollNext(output);
        long duration = System.currentTimeMillis() - startTime;
        LOG.info("KafkaSource 批量拉取并处理消息耗时 {} ms", duration);
    }

    // 其他方法直接委托给原始Reader，无需修改
    @Override
    public void start() {
        delegate.start();
    }

    @Override
    public void addSplits(List<KafkaRecordSplit> splits) {
        delegate.addSplits(splits);
    }

    @Override
    public void notifyNoMoreSplits() {
        delegate.notifyNoMoreSplits();
    }

    @Override
    public List<KafkaRecordPartitionState> snapshotState(long checkpointId) throws Exception {
        return delegate.snapshotState(checkpointId);
    }

    @Override
    public void close() throws Exception {
        delegate.close();
    }
}

补充：如果只想统计KafkaSource输出消息的时间 #

如果你的需求只是记录KafkaSource将消息输出到下游的时间，可以不用修改KafkaSource内部，直接在源之后加一个轻量的计时算子，类似peek的逻辑：

DataStream<String> sourceStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

// 添加自定义计时算子
sourceStream.transform("Kafka 输出计时", TypeInformation.of(String.class), new TimingPeekOperator<>())
        .print();

// 自定义计时算子
public class TimingPeekOperator<T> extends AbstractStreamOperator<T> implements OneInputStreamOperator<T, T> {
    private static final Logger LOG = LoggerFactory.getLogger(TimingPeekOperator.class);

    @Override
    public void processElement(StreamRecord<T> element) throws Exception {
        LOG.info("KafkaSource 输出消息，时间戳 {}，内容：{}", System.currentTimeMillis(), element.getValue());
        // 将消息传递给下游
        output.collect(element);
    }
}

内容的提问来源于stack exchange，提问作者Kleyson Rios