spring-projectsGH-3057: ReactiveKafkaConsumerTemplate missing receiveBatch method

marcusvoltolim · marcusvoltolim · commit 454a3f54bbca · 2024-02-29T15:58:02.000-03:00
* Removing `concatMap` Closes spring-projects#3057
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/reactive/ReactiveKafkaConsumerTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/reactive/ReactiveKafkaConsumerTemplate.java
@@ -72,8 +72,8 @@ public Flux<ReceiverRecord<K, V>> receive() {
 		return this.kafkaReceiver.receive();
 	}
 
-	public Flux<ReceiverRecord<K, V>> receiveBatch() {
-		return this.kafkaReceiver.receiveBatch().concatMap(Function.identity());
+	public Flux<Flux<ReceiverRecord<K, V>>> receiveBatch() {
+		return this.kafkaReceiver.receiveBatch();
 	}
 
 	public Flux<ConsumerRecord<K, V>> receiveAutoAck() {

Original file line number	Diff line number	Diff line change
`@@ -72,8 +72,8 @@ public Flux<ReceiverRecord<K, V>> receive() {`
`72`	`72`	`return this.kafkaReceiver.receive();`
`73`	`73`	`}`
`74`	`74`
`75`		`- public Flux<ReceiverRecord<K, V>> receiveBatch() {`
`76`		`- return this.kafkaReceiver.receiveBatch().concatMap(Function.identity());`
	`75`	`+ public Flux<Flux<ReceiverRecord<K, V>>> receiveBatch() {`
	`76`	`+ return this.kafkaReceiver.receiveBatch();`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`public Flux<ConsumerRecord<K, V>> receiveAutoAck() {`