Reactor kafka wrapper delegates to wrong method (#10333)

laurit · web-flow · commit bc034f3b1b8d · 2024-01-28T18:23:02.000-08:00
diff --git a/instrumentation/reactor/reactor-kafka-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactor/kafka/v1_0/InstrumentedKafkaReceiver.java b/instrumentation/reactor/reactor-kafka-1.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactor/kafka/v1_0/InstrumentedKafkaReceiver.java
@@ -59,7 +59,7 @@ public Flux<ConsumerRecord<K, V>> receiveAtmostOnce() {
   @Override
   public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(
       TransactionManager transactionManager) {
-    return actual.receiveAutoAck().map(InstrumentedKafkaReceiver::wrap);
+    return actual.receiveExactlyOnce(transactionManager).map(InstrumentedKafkaReceiver::wrap);
   }
 
   // added in 1.3.3

Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@ public Flux<ConsumerRecord<K, V>> receiveAtmostOnce() {`
`59`	`59`	`@Override`
`60`	`60`	`public Flux<Flux<ConsumerRecord<K, V>>> receiveExactlyOnce(`
`61`	`61`	`TransactionManager transactionManager) {`
`62`		`- return actual.receiveAutoAck().map(InstrumentedKafkaReceiver::wrap);`
	`62`	`+ return actual.receiveExactlyOnce(transactionManager).map(InstrumentedKafkaReceiver::wrap);`
`63`	`63`	`}`
`64`	`64`
`65`	`65`	`// added in 1.3.3`