如何在kafka流中处理给定时间范围内的key对应的最新记录？

from kafka import KafkaConsumer
from kafka import TopicPartition

def process_kafka_stream(topic, key, start_time, end_time):
    consumer = KafkaConsumer(bootstrap_servers='kafka_servers', group_id='consumer_group')
    consumer.assign([TopicPartition(topic, 0)])

    # 重置消费者偏移量为起始位置
    consumer.seek_to_beginning()

    latest_records = {}
    for message in consumer:
        record = message.value.decode('utf-8')

        # 判断记录是否在指定时间范围内
        if start_time <= record['timestamp'] <= end_time:
            if record['key'] in latest_records:
                # 更新最新记录
                if record['timestamp'] > latest_records[record['key']]['timestamp']:
                    latest_records[record['key']] = record
            else:
                latest_records[record['key']] = record

    # 输出最新记录
    for key, record in latest_records.items():
        print(f"Key: {key}, Latest Record: {record}")

    consumer.close()

# 调用函数，传入相应参数
process_kafka_stream('topic_name', 'desired_key', '2022-01-01', '2022-01-31')