search

RuckusWirelessIL / pentaho-kafka-consumer

Apache Kafka consumer step plug-in for Pentaho Kettle
Apache License 2.0
66 stars 40 forks source link

Kafka reprocess messages #20

Closed adelaocampos closed 7 years ago

adelaocampos commented 7 years ago

I am using the Kafka Consumer Plugin for Pentaho CE and would appreciate your help in its usage. I would like to know if any of you were in a situation where pentaho failed and you lost any messages (based on the official docs there's no way to read the message twice, am I wrong ?). If this situation occurs how do you capture these messages so you can reprocess them?

spektom commented 7 years ago

Kafka Consumer plug-in implements so called "high level" consumer. That means that latest offsets will be committed periodically according to the setting "auto.commit.interval.ms". What happens if a process fails? After a process will be restarted it will start consuming Kafka messages from the last committed offset, which means the same messages might be re-processed again. In short, no messages will be lost given there's enough retention in Kafka, but there might be some duplication. I suggest you to read this document for better understanding Kafka delivery semantics: https://kafka.apache.org/08/documentation.html#semantics

In order to provide "exactly once" semantics, the plug-in must be based on Kafka simple consumer, so we could better manage what offsets to start consuming from.

virivigio commented 3 years ago

Thank you @spektom for this explanation. Let me explore some details for my specific scenario: kafka_consumer -> transformation_executor(called every 1000 rows) What I see is that kafka_consumer outputs up to 10000 rows in the intra-step buffer ("Nr of rows in rowset" in transformation miscellaneous properties) and then waits. The called-transformation processes 1000 record and when it finishes another 1000 are fetched from kafka_consumer and so on. The speed -as always- depends on the slower step. Let's say that the inner transformation performs 200rows/second. So every execution lasts 5 seconds (consuming 1000 rows we said). On average. My point is: if the transformation crashes I need to replay the 10k messages in buffer. So their offset should not be committed. How can I achieve that? May I set the "auto.commit.interval.ms" to 20.000? I mean, after reading some data wait 20 seconds before committing THAT offset, so that, if you are still alive after 20s, I can assume that the inner trasformation (that should take 5seconds) is over. But, and this is my question, if this property means that every 20s the client commits the newest offset it has in memory (aka in the buffer) there is no way to replay messages upon crash because those in the buffer are committed but never get sent to the rest fo the pipeline. Have I made myself clear?

Thank you for your work and support Virgilio

spektom commented 3 years ago

Hi Virgilio,

I believe I answered you in #34 already :) Yes, you can play with auto.commit.interval.ms parameter, and set it even to 3000.

This is the downside of Kafka high-level consumer.