Offset management with Kafka

[hugocatlas]

Hello Proton team,

I'm wondering how Proton handles offsets in Kafka. Is autocommiting something that is feasible?
I see that ClickHouse handles this nicely and I just wondered whether Proton follows the same pattern.

Cheers,

Hugo

[zliang-min]

Hi Hugo,

Proton does not use consumer groups for consuming messages from Kafka topics. Instead, it handles offsets with checkpoints.

When creating a materialized view to read data from a Kafka external stream, there is a settings called checkpoint_interval which allows you to decides how often the checkpointing for that MV happens, its default value is 15 minutes. An example for setting the checkpoint interval to 60 seconds:

CREATE MATERIALIZED VIEW my_view AS SELECT * FROM my_external_stream SETTINGS checkpoint_interval=60;

Then, when checkpoint is triggered, proton will save the offsets for each partition it has consumed into the checkpoint store. And if the MV gets interrupted, for instance, proton restarts, the MV will firstly fetch the offsets from the checkpoint store and start read messages from those offsets + 1.

Also, it's possible to specify from which offsets the query should query from with the seek_to setting. There are a few formats supported by seek_to:

• latest and earliest - read from the end / beginning of the partitions. And latest is the default value for seek_to. To read from the beginning: SELECT * FROM my_ex_stream SETTINGS seek_to='earliest'.
• Offset numbers - directly specify the offsets you want to read from. E.g. SELECT * FROM my_ex_stream SETTINGS seek_to='10' will read from the offset 10. However, if a topic has multiple partitions, it's likely that each parition has different offsets, in this case, you can specify mutiple offsets in seek_to, like this: SELECT * FROM my_ex_stream SETTINGS seek_to='10,20,30'. This example reads data from a 3-partition topic, and it reads partition-0 from offset 10, partition-1 from offset 20, and partition-2 from offset 30.
• Use timestamps like 2024-04-03T20:03:00 - proton will first fetch the earliest offset whose timestamp is greater than or equal to the given timestamp for each partition, and then use the offsets to read messages. E.g. SELECT * FROM my_ex_stream SETTINGS seek_to='2024-04-03T20:03:00'. And as you might guess, it's possible to sepcify different timestamps for parititions, like SELECT * FROM my_ex_stream SETTINGS seek_to='2024-04-03T20:03:00,2024-04-03T20:04:00,2024-04-03T20:05:00'.

I hope this answers your question.

[zliang-min]

That's a fair point, Hugo. It's important to monitor the performance metrics. In terms of monitoring consumption latency, while I agree that it's easy to monitor on Kafka side with consumer groups, it's also possible to be done on Proton side. One can query the Kafka messages' timestamps with the virtual column _tp_time, i.e. SELECT _tp_time, ... FROM my_ex_stream. So the consumption latency could be monitored by comparing the _tp_time and the time when the message was read into Proton, could be something like date_diff('us', _tp_time, now()). We also provide metrics on external streams for monitoring their performance/status.

[chenziliang]

The core challenge by using consumer group is data loss: when there is complex join, aggregation and then emit to down stream system, we will need be very careful to commit / ack the offset back to Kafka customer group and for complex pipeline m, it is very hard to carry around this information through the execution pipeline.

For simple transformation without join, aggregation etc, then it is easier.

[hugocatlas]

Just wondering: would it be a possible solution to have a (internally) separate data structure that can handle consumer groups, but that cannot be used in complex transformations (join, aggregations, etc)?

I mean, when one defines the EXTERNAL STREAM, if settings like autocommit and/or consumer.group.id are present, if this stream is used downstream on JOIN or GROUP BY operations an exception is raised.

I understand that this might be out of the vision you might have for Proton. However, this would fill a gap that currently is not filled by any solution I know of (Flink, Spark, Arroyo, ...). As far as I know, the sole solution I found to handle consumer groups is FastStream, which uses internally the Kafka Python API. Well, in this case, I would rather implement it myself rather than depend on yet another Python package...

As a heavy ClickHouse user, I would be pleased to have the streaming engine able to easily horizontally scale and also have a common core with ClickHouse. This would simplify my pipelines considerably...

Anyway, I'll follow up Proton's developments and if this feature is released some when in the future, I won't hesitate to consider using it.

Cheers,

Hugo

[chenziliang]

Thanks @hugocatlas for your suggestion which is worth exploring definitely. May I ask what is your major use cases by using streaming capabilities with ClickHouse ? A horizontally scaled streaming ETL processor ? How many topics / partitions you like to process in one consumer group usually ? Have you tried ClickHouse Kafka table engine to fulfill what you like to do ?

Appreciate your feedbacks again !

[hugocatlas]

Where I work, we have several ETL pipelines that currently run on Spark Structured Streaming (and 2 other ones that we have implemented ourselves the consumer) but we have reached a consensus that we are not fully leveraging the horizontal scaling capabilities our k8s cluster offers us (i.e., KEDA). We have agreed internally that it is much better to have a variable number of very small pods instead of having a big Spark driver, even though it can spawn executor pods. In other words: Spark is overkill for our current use case. All this data land on different sinks (Elastic, S3, different ClickHouse instances). We have tried connecting ClickHouse directly to Kafka, and it works just fine. The problem is that I'm not fully comfortable with the idea of displacing to ClickHouse the responsibility to do data ingestion. Also, it doesn't solve the problem with the other sinks.

In brief, when I first read "Proton is a single binary (<500MB). No JVM or any other dependencies." I just thought: Eureka! I can already see several small pods running it on scale.