Transactions on Kafka are blocked when using instrumentation for Reactor Kafka receiver with receiveExactlyOnce

[lukny]

Describe the bug

Transactions on Kafka are blocked when using kafkaReceiver.receiveExactlyOnce(transactionManager) from Reactor Kafka Reciever with instrumentation attached from Java agent.
Temporary workaround is to use another method signature with two arguments kafkaReceiver.receiveExactlyOnce(transactionManager, null)__

Steps to reproduce

1. Setup Java project with library io.projectreactor.kafka:reactor-kafka:1.3.22and Apache Kafka
2. Configure KafkaSender and KafkaReceiver instances
3. Use receiveExactlyOnce(transactionManager) method to create transactions between consumer and producer, for example:

public void startConsuming() {
        kafkaReceiver
            .receiveExactlyOnce(kafkaProducer.transactionManager())
            .concatMap(consumerRecordFlux -> kafkaProducer.send(
                    handleMessages(consumerRecordFlux).map(record -> SenderRecord.create(record, record.key())))
                .concatWith(kafkaProducer.transactionManager().commit())
            )
            .onErrorResume(t -> kafkaProducer.transactionManager().abort().then(Mono.error(t)))
            .subscribe();
    }

Expected behavior

Transaction between consumer and producer works uninterrupted by instrumentation when using receiveExactlyOnce(transactionManager).

Actual behavior

Consumer and producer transactions cannot start because the instrumentation code blocks it.

Javaagent or library instrumentation version

at least higher than 1.30, Javaagent

Environment

JDK: 17 and 21
OS: MacOS Sonoma

Additional context

opentelemetry-java-instrumentation/instrumentation/reactor/reactor-kafka-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactor/kafka/v1_0/InstrumentedKafkaReceiver.java

Line 60 in e7f1aff

public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(

Implementation of method

public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(TransactionManager transactionManager) {
    return actual.receiveAutoAck().map(InstrumentedKafkaReceiver::wrap);
  }

uses receiveAutoAck() instead of receiveExactlyOnce()