Apache Kafka: Cum setezi offset-urile la un moment fix

  • 1 minute de citit
  • 2 exemple de cod în Python
  • 1 imagine

Disponibil și în English

miniatură blog

Salut 👋,

Acesta este un articol scurt despre setarea offset-urilor în Apache Kafka pentru un consumer group.

În mod normal, ca să resetezi offset-urile în Kafka trebuie să folosești instrumentul kafka-consumer-groups.sh, ceea ce înseamnă să descarci arhiva zip cu codul sursă Kafka și să configurezi Java SDK. Toate uneltele Kafka depind de Java, iar asta nu este prea plăcut sau prietenos pentru dezvoltatori…

Uneori, chiar reușind să pui Java la punct și să rulezi uneltele, ele nu funcționează 🤷🏻‍♂️. Fie versiunile uneltelor sunt incompatibile cu versiunea Kafka de pe server, fie comanda se execută cu succes, dar pare că nu face nimic…

O altă metodă de a seta offset-urile pentru un consumer este să folosești o bibliotecă Kafka și să faci asta prin cod.

Am Python instalat pe sistemul meu și tot ce trebuie să fac este să instalez biblioteca confluent-kafka:

pip install confluent-kafka

Apoi rulez următorul fragment de cod pentru a reseta offset-urile consumer-ului la un timestamp specific:

from confluent_kafka import Consumer, TopicPartitionimport time# Configurareconsumer_config = {    'bootstrap.servers': 'localhost:9092',    'group.id': 'my-consumer-group',    'auto.offset.reset': 'earliest',    'enable.partition.eof': True}topic = 'my-topic'timestamp_ms = int(time.mktime(time.strptime("2025-04-01 12:00:00", "%Y-%m-%d %H:%M:%S")) * 1000) # sau timpul în milisecunde# Creează consumer-ulconsumer = Consumer(consumer_config)# Obține metadatele pentru a descoperi partițiilemetadata = consumer.list_topics(topic)partitions = [TopicPartition(topic, p.id, timestamp_ms) for p in metadata.topics[topic].partitions.values()]# Caută offset-urile pentru timestampoffsets = consumer.offsets_for_times(partitions, timeout=10.0)# Atribuie partițiile cu offset-urile corecteconsumer.assign(offsets)# Începe consumultry:    while True:        msg = consumer.poll(timeout=1.0)        if msg is None:            continue        if msg.error():            print("Error:", msg.error())            continue        print(f"{msg.topic()} [{msg.partition()}] at offset {msg.offset()}: {msg.value().decode('utf-8')}")        breakexcept KeyboardInterrupt:    passfinally:    consumer.close()

Atât, mulțumesc că ai citit!