GH-2240: Bug fix for KafkaTemplate.receive(..)

Fixes GH-2240 (#2240)

- Ensures that we dont get ConsumerRecords with nulls in the list
- Ensures that we do not allow null offsets

Author: gurpiarbassi

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -70,7 +70,6 @@
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
-
 /**
  * A template for executing high-level operations. When used with a
  * {@link DefaultKafkaProducerFactory}, the template is thread-safe. The producer factory
@@ -88,6 +87,7 @@
  * @author Endika Gutierrez
  * @author Thomas Strauß
  * @author Soby Chacko
+ * @author Gurps Bassi
  */
 public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,
 		ApplicationListener<ContextStoppedEvent>, DisposableBean {
@@ -570,8 +570,14 @@ public ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested,
 		Map<TopicPartition, List<ConsumerRecord<K, V>>> records = new LinkedHashMap<>();
 		try (Consumer<K, V> consumer = this.consumerFactory.createConsumer(null, null, null, props)) {
 			requested.forEach(tpo -> {
+				if (tpo.getOffset() == null || tpo.getOffset() < 0) {
+					throw new KafkaException("Offset supplied in TopicPartitionOffset is invalid: " + tpo);
+				}
 				ConsumerRecord<K, V> one = receiveOne(tpo.getTopicPartition(), tpo.getOffset(), pollTimeout, consumer);
-				records.computeIfAbsent(tpo.getTopicPartition(), tp -> new ArrayList<>()).add(one);
+				List<ConsumerRecord<K, V>> consumerRecords = records.computeIfAbsent(tpo.getTopicPartition(), tp -> new ArrayList<>());
+				if (one != null) {
+					consumerRecords.add(one);
+				}
 			});
 			return new ConsumerRecords<>(records);
 		}

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -71,6 +71,9 @@
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
@@ -104,6 +107,7 @@
  * @author Endika Gutierrez
  * @author Thomas Strauß
  * @author Soby Chacko
+ * @author Gurps Bassi
  */
 @EmbeddedKafka(topics = { KafkaTemplateTests.INT_KEY_TOPIC, KafkaTemplateTests.STRING_KEY_TOPIC })
 public class KafkaTemplateTests {
@@ -163,6 +167,12 @@ void testTemplate() {
 
 		template.setDefaultTopic(INT_KEY_TOPIC);
 
+		template.setConsumerFactory(
+				new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
+		ConsumerRecords<Integer, String> initialRecords =
+				template.receive(Collections.singleton(new TopicPartitionOffset(INT_KEY_TOPIC, 1, 1L)));
+		assertThat(initialRecords).isEmpty();
+
 		template.sendDefault("foo");
 		assertThat(KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC)).has(value("foo"));
 
@@ -201,8 +211,7 @@ void testTemplate() {
 		assertThat(partitions).isNotNull();
 		assertThat(partitions).hasSize(2);
 		assertThat(KafkaTestUtils.getPropertyValue(pf.createProducer(), "delegate")).isSameAs(wrapped.get());
-		template.setConsumerFactory(
-				new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
+
 		ConsumerRecord<Integer, String> receive = template.receive(INT_KEY_TOPIC, 1, received.offset());
 		assertThat(receive).has(allOf(keyValue(2, "buz"), partition(1)))
 				.extracting(ConsumerRecord::offset)
@@ -621,4 +630,22 @@ void testCompositeProducerInterceptor() {
 		inOrder.verify(producerInterceptor2).onAcknowledgement(any(RecordMetadata.class), Mockito.isNull());
 	}
 
+	@ParameterizedTest(name = "{0} is invalid")
+	@NullSource
+	@ValueSource(longs =  -1)
+	void testReceiveWhenOffsetIsInvalid(Long offset) {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+
+		template.setConsumerFactory(
+				new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
+		TopicPartitionOffset tpoWithNullOffset = new TopicPartitionOffset(INT_KEY_TOPIC, 1, offset);
+
+		assertThatExceptionOfType(KafkaException.class)
+				.isThrownBy(() -> template.receive(Collections.singleton(tpoWithNullOffset)))
+				.withMessage("Offset supplied in TopicPartitionOffset is invalid: " + tpoWithNullOffset);
+	}
+
+
 }