Kafka基准测试中延迟线性增长、生产者缓冲区堆积问题咨询

项目背景 #

当前正在开展Apache Kafka消息传输延迟的基准测试工作：

• 初始测试数据集为10MB推特数据（约14500条推文），后续可根据测试需求调整数据发送规模
• Broker配置为使用LogAppendTime作为消息自带时间戳
• 延迟计算逻辑：在Consumer侧提取Broker写入的LogAppendTime时间戳，与消费消息时的本地系统时间做差，得到消息端到端传输延迟
• 实际观测到的延迟值随运行时间持续升高，与预期结果存在明显偏差

测试环境配置 #

• 运行环境：Ubuntu 20 虚拟机
• 虚拟机资源配置：2 VPU、9GB内存
• 组件版本：最新版confluent-kafka 与 Apache Kafka
• 集群与Topic配置：单Kafka Broker、Topic仅设置1个分区，除LogAppendTime参数外其余配置全部保持默认
• 部署位置：生产者、消费者均运行在同一台虚拟机上，虚拟机部署在数据中心内部，理论上不存在网络带宽瓶颈

固定常量配置 #

CONSUMER_GROUP_ID = "twitter-consumers"
KAFKA_TOPIC_TWITTER = "twitter-stream"

核心实现代码 #

生产者实现 #

producer_config = {
    "bootstrap.servers": "localhost:9092"
}
p = Producer(producer_config)

generation_steps = 1 # 10 MBs per step
with open(TWITTER_DATA_PATH, "r") as dataset:
    for step in range(generation_steps):
        print(f"Executing data generation step {step}...")
        dataset.seek(0) # Jump back to first line  
 
        for tweet in dataset:
            try:
                print("IN QUEUE: {}".format(len(p)))
                p.produce(KAFKA_TOPIC_TWITTER, value=tweet)
                p.poll(0)
            except BufferError:
                print('[INFO] Local producer queue is full (%d messages awaiting delivery): Trying again after flushing...\n' % len(p)) 
                p.flush() # Send the messages in the queue

                # The tweet that caused the BufferError could not be sent, so after flushing, we can now send it
                p.produce(KAFKA_TOPIC_TWITTER, value=tweet)
                p.poll(0)

print("Data generation done!")

消费者实现 #

# Consume the data
consumer_config = {
    "bootstrap.servers": "localhost:9092",
    "group.id": CONSUMER_GROUP_ID,
    "client.id": "client-1",
    "enable.auto.commit": True,
    "session.timeout.ms": 6000,
    "default.topic.config": {"auto.offset.reset": "smallest"}
}
c = Consumer(consumer_config)

c.subscribe([KAFKA_TOPIC_TWITTER])

msg_count = 0
try:
    while True:
        msg = c.poll(0.1)
        
        if msg is None:
            continue
        elif not msg.error():
            delta = (time.time() * 1000) - msg.timestamp()[1]
            msg_count += 1
            print("Message {} received | Timestamp: {} | Latency: {}".format(msg_count, msg.timestamp()[1], delta))
        elif msg.error().code() == KafkaError._PARTITION_EOF:
            print("End of partition reached {}/{}".format(msg.topic(), msg.partition()))
        else:
            print("Error occured: {}".format(msg.error().str()))
except KeyBoardInterrupt:
    pass
finally:
    c.close()

实际观测结果 #

消费者侧延迟观测 #

Message 1 received | Timestamp: 1618685061270 | Latency: 12.360107421875
Message 2 received | Timestamp: 1618685061270 | Latency: 13.025390625
Message 3 received | Timestamp: 1618685061270 | Latency: 13.08544921875
Message 4 received | Timestamp: 1618685061270 | Latency: 13.156005859375
Message 5 received | Timestamp: 1618685061270 | Latency: 13.1923828125
Message 6 received | Timestamp: 1618685061270 | Latency: 13.21875
Message 7 received | Timestamp: 1618685061270 | Latency: 13.245849609375
Message 8 received | Timestamp: 1618685061270 | Latency: 13.29833984375
Message 9 received | Timestamp: 1618685061270 | Latency: 13.348388671875
Message 10 received | Timestamp: 1618685061270 | Latency: 13.389892578125
Message 11 received | Timestamp: 1618685061270 | Latency: 13.43701171875
Message 12 received | Timestamp: 1618685061270 | Latency: 13.49755859375
Message 13 received | Timestamp: 1618685061270 | Latency: 13.54345703125
Message 14 received | Timestamp: 1618685061270 | Latency: 13.581787109375
Message 15 received | Timestamp: 1618685061270 | Latency: 13.626708984375
Message 16 received | Timestamp: 1618685061270 | Latency: 13.661376953125
Message 17 received | Timestamp: 1618685061270 | Latency: 13.695556640625
Message 18 received | Timestamp: 1618685061270 | Latency: 13.720703125
Message 19 received | Timestamp: 1618685061270 | Latency: 13.75537109375
Message 20 received | Timestamp: 1618685061270 | Latency: 13.78857421875
Message 21 received | Timestamp: 1618685061270 | Latency: 13.81396484375
[....]
Message 14485 received | Timestamp: 1618685061337 | Latency: 1061.004150390625

生产者侧队列堆积观测 #

Executing data generation step 0...
IN QUEUE: 0
IN QUEUE: 1
IN QUEUE: 2
IN QUEUE: 3
IN QUEUE: 4
IN QUEUE: 5
IN QUEUE: 6
IN QUEUE: 7
IN QUEUE: 8
IN QUEUE: 9
IN QUEUE: 10
IN QUEUE: 11
IN QUEUE: 12
IN QUEUE: 13
IN QUEUE: 14
IN QUEUE: 15
IN QUEUE: 16
[...]
IN QUEUE: 14484
Data generation done!

待排查的核心问题 #

通过delta = (time.time() * 1000) - msg.timestamp()[1]计算消息延迟，原本预期延迟值会在10-100ms区间波动，但实际观测到延迟随消费进度呈线性增长；同时按照固有认知，Kafka调用.poll()方法后应该清空本地发送队列，但实际生产过程中生产者本地队列堆积到14000+条消息，需要定位异常诱因与对应的解决方法。

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