Flink消费Kafka消息速度慢且存在滞后问题排查求助

问题现象 #

• 高吞吐场景：每秒向Kafka生产130万条消息时，Flink消费者每秒仅处理约120万条，records-lag-max指标从数十万持续增长至超百万
• 低吞吐场景：将生产速率降至每秒10万条后，资源配置不变，但仍存在约1万条消息滞后，不符合预期

测试用Flink作业代码 #

env.setParallelism(25);
env.setMaxParallelism(25);

........

var ssgKafkaSource = SlotSharingGroup.newBuilder("kafka source")
                .setCpuCores(5.3)
                .setTaskHeapMemoryMB(19666)
                .build();

var deserializationSchema = ConfluentRegistryAvroDeserializationSchema.forSpecific(Message.class, jobProperties.schemaRegistryUrl);

KafkaSource<Message> source = KafkaSource.<Message>builder()
        .setBootstrapServers(jobProperties.bootstrapServers)
        .setGroupId(jobProperties.consumerGroupId)
        .setTopics(jobProperties.inputTopic)
        .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
        .setValueOnlyDeserializer(deserializationSchema)
        .build();
         
DataStream<Message> messages = env
                    .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
                    .slotSharingGroup(ssgKafkaSource);
    
var messages = messages.map(m -> 1);

已完成的排查与优化 #

• 序列化：遵循Flink序列化调优指南，使用Flink Avro序列化器，引入flink-avro依赖，Message类继承SpecificRecordBase
• 运行环境：Task Manager部署在配备32vCores、256GB内存及低延迟SSD的EC2实例上，已排除RocksDB磁盘相关性能问题
• Kafka参数调优：尝试调整fetch.max.bytes、max.poll.records等消费者属性，未观察到明显效果
• 并行度匹配：Kafka主题含25个分区，与Flink作业设置的并行度（25）保持一致

求排查遗漏的性能瓶颈点

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