Sending several messages in a single transaction and reading the message headers

[Sgitario]

I have the following use case:

1. consume one message from a channel and read some headers from the incoming message.
2. perform some database queries
3. emit several messages to another channel as a result of the database queries

All the above steps need to be executed in a single transaction, so for example, if the database fails, the message is nack, or if the emission of the messages fails the database queries are rollback.

What I've tried so far:

@ApplicationScoped
public class Consumer {
  public static final String RECEIVED_KEY = "kafka_receivedMessageKey";

  @Channel("billable-usage-out") KafkaTransactions<BillableUsage> channel;

  @Incoming("tally-summary")
  @Transactional
  public Uni<Void> consume(Message<Payload> message) {
    MessageHelper.findFirstHeaderAsString(message, RECEIVED_KEY)
        .ifPresent(
            kafkaMessageKey ->
                log.debug(
                    "Tally Summary received w/ key={}. Producing billable usage.",
                    kafkaMessageKey));

    return channel.withTransactionAndAck(message, emitter -> {
      // perform database queries and updates
      doSomethingWithThePayload(message.getPayload);
      // emit several messages
      emitter.send(x);
      emitter.send(y);
      // ...
      return Uni.createFrom().voidItem();
    });
  }

This fails with the following exception:

2024-05-07 17:21:35,001 ERROR [io.sma.rea.mes.provider] (executor-thread-2) SRMSG00200: The method com.redhat.swatch.billable.usage.services.TallySummaryMessageConsumer#consume has thrown an exception: java.lang.IllegalStateException: SRMSG18018: Unable to find the Kafka consumer for channel `tally-summary`
	at io.smallrye.reactive.messaging.kafka.transactions.KafkaTransactionsImpl.withTransaction(KafkaTransactionsImpl.java:102)
	at io.smallrye.reactive.messaging.kafka.transactions.KafkaTransactions.withTransactionAndAck(KafkaTransactions.java:120)
	at com.redhat.swatch.billable.usage.services.Consumer.consume(Consumer.java:76)

So, for some reason, the channel from the incoming message is not found as a consumer.

I also tried the following without luck:

@Incoming("tally-summary")
  @Outgoing("billable-usage-out")
  @Transactional
  public Multi<BillableUsage> consume(Message<Payload> message) {
    MessageHelper.findFirstHeaderAsString(message, RECEIVED_KEY)
        .ifPresent(
            kafkaMessageKey ->
                log.debug(
                    "Tally Summary received w/ key={}. Producing billable usage.",
                    kafkaMessageKey));

    return Multi.createFrom().items(service.fromTallySummary(message.getPayload())
            .map(billableUsageService::submitBillableUsage));
  }

In this case, this fails because we can't get Message<Tally> but Multi, but using Multi, we can't obtain the message headers anyway, right? Yet, I'm not sure if the transaction would work.

Can somebody help about how I can do the above?

[Sgitario]

maybe @ozangunalp can help with this?

[ozangunalp]

I'll check the problem with KafkaTransactions, it shouldn't happen.

Without transactions, normally you can return a Multi<Payload> when you consume a payload too :

  @Incoming("tally-summary")
  @Outgoing("billable-usage-out")
  @Transactional
  public Multi<BillableUsage> consume(Tally payload, IncomingKafkaRecordMetadata metadata) {
    // Instead of message you use the metadata to extract the header
    MessageHelper.findFirstHeaderAsString(metadata, RECEIVED_KEY)
        .ifPresent(
            kafkaMessageKey ->
                log.debug(
                    "Tally Summary received w/ key={}. Producing billable usage.",
                    kafkaMessageKey));

    return Multi.createFrom().items(service.fromTallySummary(payload)
            .map(billableUsageService::submitBillableUsage));
  }

[Sgitario]

I didn't know we can inject the metadata like this public Multi<BillableUsage> consume(Tally payload, IncomingKafkaRecordMetadata metadata) {, thanks for pointing it out.

I forgot to mention that I'm also using Hibernate ORM, so the consumer needs to be annotated with @Blocking.

For example, doing:

@Channel("billable-usage-out")
  KafkaTransactions<Payload> channel;

  @Incoming("tally-summary")
  @Transactional
  @Blocking
  public Uni<Void> consume(Message<Payload> message) {
    MessageHelper.findFirstHeaderAsString(message, RECEIVED_KEY)
        .ifPresent(
            kafkaMessageKey ->
                log.debug(
                    "Tally Summary received w/ key={}. Producing billable usage.",
                    kafkaMessageKey));

    return channel.withTransactionAndAck(
        message,
        emitter -> {
          mapper
              .fromTallySummary(message.getPayload())
              .forEach(usage -> emitter.send(service.produceNew(usage)));
          return Uni.createFrom().voidItem();
        });
  }

Fails with:

2024-05-07 19:46:33,790 ERROR [io.sma.rea.mes.kafka] (smallrye-kafka-consumer-thread-1) SRMSG18203: A message sent to channel `tally-summary` has been nacked, fail-stop
2024-05-07 19:46:33,791 WARN  [io.sma.rea.mes.kafka] (smallrye-kafka-consumer-thread-1) SRMSG18228: A failure has been reported for Kafka topics '[platform.rhsm-subscriptions.tally]': io.quarkus.runtime.BlockingOperationNotAllowedException: You have attempted to perform a blocking operation on a IO thread. This is not allowed, as blocking the IO thread will cause major performance issues with your application. If you want to perform blocking EntityManager operations make sure you are doing it from a worker thread.
        at io.quarkus.hibernate.orm.runtime.session.TransactionScopedSession.checkBlocking(TransactionScopedSession.java:116)
        at io.quarkus.hibernate.orm.runtime.session.TransactionScopedSession.getCriteriaBuilder(TransactionScopedSession.java:537)
        at org.hibernate.engine.spi.SessionLazyDelegator.getCriteriaBuilder(SessionLazyDelegator.java:699)
        at org.hibernate.engine.spi.SessionLazyDelegator.getCriteriaBuilder(SessionLazyDelegator.java:67)
        at org.hibernate.Session_OpdLahisOZ9nWRPXMsEFQmQU03A_Synthetic_ClientProxy.getCriteriaBuilder(Unknown Source)
        at com.redhat.swatch.billable.usage.data.BillableUsageRemittanceRepository.getRemittanceSummaries(BillableUsageRemittanceRepository.java:50)

Because service.produceNew(...) needs the entity manager.

Honestly, I'm a bit blocked and I would really appreciate any hints :)

[Sgitario]

Finally, I scaled up the issue in the the Quarkus repo since I'm not sure if this is related to the Hibernate ORM extension or the fix needs to be aligned with it. quarkusio/quarkus#40530

[ozangunalp]

Sorry I couldn't respond earlier.
I don't think you can chain currently Kafka transactions with Hibernate ORM but you can do it using Hibernate Reactive. Because Kafka transactions run on the event loop and Hibernate ORM is blocking.

I am not sure how to address this limitation.
Each time you use KafkaTransactions, withTransaction code is wrapped around begin/commit transaction calls to the producer (and more if you are doing exactly-once processing), and to preserve the Vert.x context we are running the emitter code if we capture a Vert.x context on the caller (for cases of @Blocking a vert.x worker thread).