Kafka Compact Topic read all events and move onto next action

[cookie-bytes]

Hi

We have a use case where we are using Kafka compact topics.
We need to read all these events on the compact topic, and only when we get to the end of the log can we proceed to the next part of the function to run.

Can this be achieved in Silverback?

Thanks

[msallin]

We have such a scenario in which we use Kafka as a database i.e. the service starts and we have to wait for all records to be consumed. Conceptually, you have to ensure that you also compact in memory to ensure you have the current view onto the topic. The reason is that Kafka does not guarantee a point in time for compaction, means you can still receive two messages with the same Kafka key (i.e. for the same entity) on startup. However, there is ofc also the scenario that you receive a new message after the startup.
The good news is that you can just unify this to one concept and you are able to perform the initial load and then also have a constantly updated view to the topic.

We have

1. multiple classes (one per entity) registered as singleton. Actually in memory repositories.
2. a health check gets the repositories injected, they expose a "InitialLoadCompleted" flag. If this is true, the health check returns Healthy.
3. a IKafkaPartitionEofCallback that knows when end of partition is reached and sets flag on the corresponding repository (register it like this)
4. a KafkaSubscriber per entity/topic that has the methods ExecuteAsync(Entity e) and DeleteAsync(Tombstone<Entity> t), the ExecuteAsync calls AddOrUpdate on the repository and the DeleteAsync delete (using the MessageId given from t).
5. a configuration on the input endpoints to always consume from the beginning and to activate EoF notifications.

.ConsumeFrom(new TopicPartitionOffset(new TopicPartition("MyTopic", 0), Offset.Beginning))
.Configure(config =>
{
    config.EnablePartitionEof = true;
})

How this works together: Kubernetes will not expose the API and does show the deployment in progress as long as the readiness check isn't healthy. As soon as the application starts, the partitions of the topic are consumed from the beginning. As long as the end of partition message isn't sent by Kafka, the health check returns unhealthy. The message are consumed from the Subscriber as if they would in a normal scenario and the in memory DB (actually a concurrent dictionary) is populated. When all message were consumed Kafka sends the end of partition message and the corresponding handler sets the flag on the repository. When this happened for all topics, the health check will return healthy and traffic is routed to the API.

If you don't use Kubernetes and thus the health check isn't an option you have to get a bit more creative. I would do something like this: Do the same expect point 2 (the health check). Instead exposing a flag from the repositories, expose a task. Then, inject the repository in that place you want to wait for completion. The method call on the handler completes the task instead of setting a flag to true. Voilà.

@BEagle1984 we could massively simplify that. E.g. on the IBroker we already expose the consumers. There could be such a task that is populated when EnablePartitionEof is enabled.

[BEagle1984]

@cookie-bytes you can use the EOF callback as explained by @msallin (thanks bro!): see Kafka Events and specifically IKafkaPartitionEofCallback.

The idea is that you get notified when you consumed the whole partition, but it's up to you to build the rest of the logic (i.e. triggering the next step only when all partitions reach the EOF etc.).

Yes @msallin, in the future I'd like to offer a better abstraction for this, but I'm not sure about the API. I would probably just build a sort of "topic EOF" callback that fires when all partitions reach EOF, for homogeneity.

[cookie-bytes]

The topic EOF callback makes sense for future enhancement.

[cookie-bytes]

Thank you both