GH-1693: Use concatMap in ReactiveKafkaTemplate

oemergenc · web-flow · commit 9101166b7eae · 2021-01-27T10:43:05.000-05:00
Fixes #1693 The documentation of the `reactor-kafka` project states that messages should be consumed using the `concatMap` operator, but the `ReactiveKafkaConsumerTemplate` uses the `flatMap` operator at least in case of the `receiveAutoAck` method.
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
@@ -66,7 +66,7 @@ public Flux<ReceiverRecord<K, V>> receive() {
 	}
 
 	public Flux<ConsumerRecord<K, V>> receiveAutoAck() {
-		return this.kafkaReceiver.receiveAutoAck().flatMap(Function.identity());
+		return this.kafkaReceiver.receiveAutoAck().concatMap(Function.identity());
 	}
 
 	public Flux<ConsumerRecord<K, V>> receiveAtMostOnce() {

Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ public Flux<ReceiverRecord<K, V>> receive() {`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`public Flux<ConsumerRecord<K, V>> receiveAutoAck() {`
`69`		`- return this.kafkaReceiver.receiveAutoAck().flatMap(Function.identity());`
	`69`	`+ return this.kafkaReceiver.receiveAutoAck().concatMap(Function.identity());`
`70`	`70`	`}`
`71`	`71`
`72`	`72`	`public Flux<ConsumerRecord<K, V>> receiveAtMostOnce() {`