Creating an at-least-once Kafka producer on multiple application instances.

[hans-drexler-crv4all-com]

We want to send a batch of message to a dynamic topic. And we want to be sure that a message arrives at least once.
We would rather not have a ack for every message, for performance considerations.

Our current solution (code below) works as long as there is a single application instance, but fails when there are more instances. In that case this error message appears:

Kafka producer failed: org.apache.kafka.common.errors.ProducerFencedException: Producer with transactionalId 'joindata-outbox' and (producerId=399747322, epoch=22) has been fenced by another producer with the same transactionalId

Our theory is that every instance should set its own unique transactional-id. We tried this, but are not able to get it working:

Adding the line below to application.properties has no effect. We still get the producer fenced exception with transactionalId 'joindata-outbox' as if the setting has no effect. Line:
mp.messaging.outgoing.outbox.transactional-id=${quarkus.application.name}-${channelName}-${quarkus.uuid}

And with this variant, it also does not work (We get an error about a non-transactional producer being used):
mp.messaging.outgoing.outbox.transactional.id=${quarkus.application.name}-${channelName}-${quarkus.uuid}

Question: How should we code a producer that:

• Ensures at-least-once delivery
• Allows us to select the topic dynamically
• Works in an environment with more than 1 producer instances

We are using Quarkus version 3.20.0. Any advise is greatly appreciated. Are we on the right track or not?

This is the code of the message sending:

    @Inject
    @Channel("outbox")
    KafkaTransactions<OutboxGenericMessage> txProducer;
    
    public void sendMessages(List<OutboxGenericMessage> messages, String entityName) {
        LOGGER.debug("Sending a chunk of messages. (size/entity name) {}/{}", messages.size(), entityName);
        try {
            Instant t1 = Instant.now();
            List<KafkaRecord<String, OutboxGenericMessage>> list = messages.stream().map(this::createRecord).toList();
            Uni<Void> result = txProducer.withTransaction(emitter -> {
                for (KafkaRecord<String, OutboxGenericMessage> record : list) {
                    emitter.send(record);
                }
                return Uni.createFrom().voidItem();
            });
            result.log().await().indefinitely();
            long duration = Duration.between(t1, Instant.now()).toMillis();
            if (duration == 0L) {
                duration = 1L;
            }
            LOGGER.info("Sent messages to Kafka. entity/count/time/speed {}/{}/{}ms/{} messages/sec",
                    entityName, messages.size(), duration, (messages.size() * 1000L) / duration);
        } catch (Exception e) {
            String exceptionMessage = e.toString();
            LOGGER.error("Kafka producer failed: {}", exceptionMessage);
            if (e.getCause() != null) {
                LOGGER.error("kafka producer error cause: {}", e.getCause().getMessage());
            }
            LOGGER.error("Kafka producer exception: ", e);
            throw new OutboxKafkaAbortException("Kafka producer failed sending messages", e);
        }
    }

application.properties:

# Kafka config
mp.messaging.connector.smallrye-kafka.bootstrap.servers=<redacted>
mp.messaging.connector.smallrye-kafka.sasl.jaas.config=<redacted>
mp.messaging.connector.smallrye-kafka.client.id=joindata-domain
mp.messaging.connector.smallrye-kafka.group.id=<redacted>
mp.messaging.connector.smallrye-kafka.max.request.size=10485760
mp.messaging.connector.smallrye-kafka.max-inflight-messages=50
mp.messaging.connector.smallrye-kafka.security.protocol=SASL_SSL
mp.messaging.connector.smallrye-kafka.sasl.mechanism=PLAIN
mp.messaging.connector.smallrye-kafka.client.dns.lookup=use_all_dns_ips
mp.messaging.connector.smallrye-kafka.session.timeout.ms=45000
mp.messaging.connector.smallrye-kafka.acks=all
mp.messaging.connector.smallrye-kafka.failure-strategy=fail
mp.messaging.connector.smallrye-kafka.schema.registry.url=<redacted>
mp.messaging.connector.smallrye-kafka.basic.auth.credentials.source=USER_INFO
mp.messaging.connector.smallrye-kafka.basic.auth.user.info=hidden:hidden
quarkus.kafka.devservices.enabled=false
quarkus.apicurio-registry.devservices.enabled=false

# kafka outbox topic
mp.messaging.outgoing.outbox.connector=smallrye-kafka
mp.messaging.outgoing.outbox.topic=outbox.helixs.joindata-domain # placeholder: topic name is dynamic: intake.helixs_<source>.<entity>
mp.messaging.outgoing.outbox.auto.register.schemas=true
mp.messaging.outgoing.outbox.key.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.outbox.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer