CODEWITHSUNDEEP
apache-kafkamicronaut
user1461555
user1461555

Reputation: 11

Micronaut kafka : unable to use exactly once kafka message semantics

I am using micronaut kafka to set up my producer. I am using the @KafkaClient annotation to set up all the producer config Micronaut kafka enables me set all the parameters to set up a transactional producer. When I push the message, i get back an exception saying

io.micronaut.messaging.exceptions.MessagingClientException: Exception sending producer record: Cannot perform 'send' before completing a call to initTransactions when transactions are enabled.

Referring back to the mirconaut documentation section looks like it is asking you to use KafkaProducer API to implement this feature.

From what I can assess KafkaProducer.initTransactions() method needs to be invoked before starting transactions and doesn't look that it is happening.

Has anyone faced a similar issue implementing this?

Upvotes: 1

Views: 685

Answers (1)

Bai X.
Bai X.

Reputation: 19

I guess, you are using single-node-cluster for development right? If it is so, you should configure transaction.state.log.min.isr=1 and transaction.state.log.replication.factor=1 on your local cluster. They are all preconfigured with 3 by default.

There is also a section from confluent https://docs.confluent.io/current/streams/developer-guide/config-streams.html

processing.guarantee
The processing guarantee that should be used. Possible values are "at_least_once" (default) and "exactly_once". Note that if exactly-once processing is enabled, the default for parameter commit.interval.ms changes to 100ms. Additionally, consumers are configured with isolation.level="read_committed" and producers are configured with retries=Integer.MAX_VALUE and enable.idempotence=true per default. Note that "exactly_once" processing requires a cluster of at least three brokers by default, which is the recommended setting for production. For development, you can change this by adjusting the broker settings in both transaction.state.log.replication.factor and transaction.state.log.min.isr to the number of brokers you want to use.

Upvotes: 1

Related Questions