How to start consuming specific partitions?

[kosa-gyula-77]

I would like to duplicate the 'partitions assigned' behaviour implemented at ExactlyOnce with kafka-flow.

One of my consumer is configured WithPartitionsAssignedHandler. In this handler I would like to programatically start another consumer to consume the assigned partitions from start to end.

I configure the other consumer WithInitialState(ConsumerInitialState.Stopped), then I Resume the assigned partitions in PartitionsAssignedHandler as illustrated below.

Is this the correct way to do it with kafka-flow, please?

public void PartitionsAssignedHandler(IDependencyResolver dependencyResolver, List<TopicPartition> partitions)
{
	using var scope = dependencyResolver.CreateScope();
	var consumerAccessor = scope.Resolver.Resolve<IConsumerAccessor>();
	var consumer = consumerAccessor["event-consumer"];

	// Rebuild the domain model for the assigned partitions from the events.
	RebuildDomainModel(partitions, consumer);
}

private void RebuildDomainModel(IReadOnlyCollection<TopicPartition> partitions, IMessageConsumer consumer)
{
	if (consumer.Status != ConsumerStatus.Stopped)
	{
		consumer.StopAsync().GetAwaiter().GetResult();
	}
	consumer.Resume(partitions);

	while (true)
	{
		Thread.Sleep(1000); // Give some time to the consumer to process messages

		if (partitions.Count(p => consumer.GetPosition(p) == Offset.End) == partitions.Count)
		{
			break;
		}
	}

	consumer.StopAsync().GetAwaiter().GetResult();
}

[filipeesch]

Hi @kosa-gyula-77, if you explain what you want to do in your application will be easier to help :)

[kosa-gyula-77]

Hi! Thanks for reaching out!

I'm implementing an event store with Kafka. The domain model is partitioned among the consumers of the commands the model accepts as input. Upon successful command consumption the domain produces event messages. The events topic has the very same number of partitions as the commands topic. The record of event messages acts as the event store.

Whenever one or more partition is assigned to a different command consumer instance, it must rebuild the corrsponding partition(s) of the domain model by replaying (consuming) the events. So, when partitions are assigned to the command consumer, I would like to consume the events that belong to the assigned partitions from beginning to end.

Confluent.Kafka has an Assign method on the IConsumer interface that I cannot find in kafka-flow. This method is used in the sample ExactlyOnce.

Does that make sense, please? Thanks!

[filipeesch]

I think a Kafka topic is not the best "storage" to use as an event store. I think a database is more appropriate to do that (or even a Kafka table, I don't know how it works).

[kosa-gyula-77]

Okay, so I assume that it is not yet supported to start a consumer to consume only select partitions from start to end.

[kosa-gyula-77]

@filipeesch FYKI It seems that Kafka is a viable product for event store according to https://www.confluent.io/blog/okay-store-data-apache-kafka/

BTW What did you mean on 'Kafka table' , please? Did you perhaps mean ksqlDB?

[filipeesch]

You can store your events there, but the query capabilities of a Kafka topic are very limited, you can only query using the topic/partition/offiset combination and AFAIK the Confluent Driver does not expose this capability, only continuous reading. If you want to query all events from an aggregate you won't be able to do it.

Yes, I mean ksqlDB, but I don't know how it works and how to use it.