GH-2673: Add Mock Consumer and Producer Factories

Resolves #2673

Also remove an unnecessary call to `offsetsForTimes` which is not supported
by `MockConsumer`.

* Doc polishing.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/testing.adoc

@@ -624,3 +624,71 @@ received = records.poll(10, TimeUnit.SECONDS);
 assertThat(received).has(allOf(keyValue(2, "baz"), partition(0)));
 ----
 ====
+
+[[mock-cons-prod]]
+==== Mock Consumer and Producer
+
+The `kafka-clients` library provides `MockConsumer` and `MockProducer` classes for testing purposes.
+
+If you wish to use these classes in some of your tests with listener containers or `KafkaTemplate` respectively, starting with version 3.0.7, the framework now provides `MockConsumerFactory` and `MockProducerFactory` implementations.
+
+These factories can be used in the listener container and template instead of the default factories, which require a running (or embedded) broker.
+
+Here is an example of a simple implementation returning a single consumer:
+
+====
+[source, java]
+----
+@Bean
+ConsumerFactory<String, String> consumerFactory() {
+    MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+    TopicPartition topicPartition0 = new TopicPartition("topic", 0);
+    List<TopicPartition> topicPartitions = Arrays.asList(topicPartition0);
+    Map<TopicPartition, Long> beginningOffsets = topicPartitions.stream().collect(Collectors
+            .toMap(Function.identity(), tp -> 0L));
+    consumer.updateBeginningOffsets(beginningOffsets);
+    consumer.schedulePollTask(() -> {
+        consumer.addRecord(
+                new ConsumerRecord<>("topic", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test1",
+                        new RecordHeaders(), Optional.empty()));
+        consumer.addRecord(
+                new ConsumerRecord<>("topic", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test2",
+                        new RecordHeaders(), Optional.empty()));
+    });
+    return new MockConsumerFactory(() -> consumer);
+}
+----
+====
+
+If you wish to test with concurrency, the `Supplier` lambda in the factory's constructor would need create a new instance each time.
+
+With the `MockProducerFactory`, there are two constructors; one to create a simple factory, and one to create factory that supports transactions.
+
+Here are examples:
+
+====
+[source, java]
+----
+@Bean
+ProducerFactory<String, String> nonTransFactory() {
+    return new MockProducerFactory<>(() -> 
+            new MockProducer<>(true, new StringSerializer(), new StringSerializer()));
+}
+
+@Bean
+ProducerFactory<String, String> transFactory() {
+    MockProducer<String, String> mockProducer = 
+            new MockProducer<>(true, new StringSerializer(), new StringSerializer());
+    mockProducer.initTransactions();
+    return new MockProducerFactory<String, String>((tx, id) -> mockProducer, "defaultTxId");
+}
+----
+====
+
+Notice in the second case, the lambda is a `BiFunction<Boolean, String>` where the first parameter is true if the caller wants a transactional producer; the optional second parameter contains the transactional id.
+This can be the default (as provided in the constructor), or can be overridden by the `KafkaTransactionManager` (or `KafkaTemplate` for local transactions), if so configured.
+The transactional id is provided in case you wish to use a different `MockProducer` based on this value.
+
+If you are using producers in a multi-threaded environment, the `BiFunction` should return multiple producers (perhaps thread-bound using a `ThreadLocal`).
+
+IMPORTANT: Transactional `MockProducer` s must be initialized for transactions by calling `initTransaction()`.

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -98,3 +98,9 @@ Four constants in `KafkaHeaders` that were deprecated in 2.9.x have now been rem
 * Instead of `PARTITION_ID`, use `PARTITION`
 
 Similarly, `RECEIVED_MESSAGE_KEY` is replaced by `RECEIVED_KEY` and `RECEIVED_PARTITION_ID` is replaced by `RECEIVED_PARTITION`.
+
+[[x30-testing]]
+==== Testing Changes
+
+Version 3.0.7 introduced a `MockConsumerFactory` and `MockProducerFactory`.
+See <<mock-cons-prod>> for more information.

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -3225,15 +3225,17 @@ private void initPartitionsIfNeeded() {
 			Map<TopicPartition, Long> times = partitions.entrySet().stream()
 					.filter(e -> SeekPosition.TIMESTAMP.equals(e.getValue().seekPosition))
 					.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().offset));
-			Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = this.consumer.offsetsForTimes(times);
-			offsetsForTimes.forEach((tp, off) -> {
-				if (off == null) {
-					ends.add(tp);
-				}
-				else {
-					partitions.put(tp, new OffsetMetadata(off.offset(), false, SeekPosition.TIMESTAMP));
-				}
-			});
+			if (!times.isEmpty()) {
+				Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = this.consumer.offsetsForTimes(times);
+				offsetsForTimes.forEach((tp, off) -> {
+					if (off == null) {
+						ends.add(tp);
+					}
+					else {
+						partitions.put(tp, new OffsetMetadata(off.offset(), false, SeekPosition.TIMESTAMP));
+					}
+				});
+			}
 			doInitialSeeks(partitions, beginnings, ends);
 			if (this.consumerSeekAwareListener != null) {
 				this.consumerSeekAwareListener.onPartitionsAssigned(this.definedPartitions.keySet().stream()

spring-kafka/src/main/java/org/springframework/kafka/mock/MockConsumerFactory.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.mock;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.MockConsumer;
+
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.lang.Nullable;
+
+/**
+ * Support the use of {@link MockConsumer} in tests.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 3.0.7
+ *
+ */
+public class MockConsumerFactory<K, V> implements ConsumerFactory<K, V> {
+
+	private final Supplier<MockConsumer> consumerProvider;
+
+	/**
+	 * Create an instance with the supplied consumer provicer.
+	 * @param consumerProvider the consumer provider.
+	 */
+	public MockConsumerFactory(Supplier<MockConsumer> consumerProvider) {
+		this.consumerProvider = consumerProvider;
+	}
+
+	@Override
+	public Map<String, Object> getConfigurationProperties() {
+		return Collections.emptyMap();
+	}
+
+	@Override
+	public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
+			@Nullable String clientIdSuffix) {
+
+		return this.consumerProvider.get();
+	}
+
+	@Override
+	public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
+			@Nullable String clientIdSuffix, @Nullable Properties properties) {
+
+		return this.consumerProvider.get();
+	}
+
+	@Override
+	public boolean isAutoCommit() {
+		return false;
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/mock/MockProducerFactory.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.mock;
+
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
+
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.lang.Nullable;
+
+/**
+ * Support the use of {@link MockProducer} in tests.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 3.0.7
+ *
+ */
+public class MockProducerFactory<K, V> implements ProducerFactory<K, V> {
+
+	private final BiFunction<Boolean, String, MockProducer<K, V>> producerProvider;
+
+	@Nullable
+	private final String defaultTxId;
+
+	private final boolean transactional;
+
+	/**
+	 * Create an instance that does not support transactional producers.
+	 * @param producerProvider a {@link Supplier} for a {@link MockProducer}.
+	 */
+	public MockProducerFactory(Supplier<MockProducer> producerProvider) {
+		this.producerProvider = (tx, id) -> producerProvider.get();
+		this.defaultTxId = null;
+		this.transactional = false;
+	}
+
+	/**
+	 * Create an instance that supports transactions, with the supplied producer provider {@link BiFunction}. The
+	 * function has two parameters, a boolean indicating whether a transactional producer
+	 * is being requested and, if true, the transaction id prefix for that producer.
+	 * @param producerProvider the provider function.
+	 * @param defaultTxId the default transactional id.
+	 */
+	public MockProducerFactory(BiFunction<Boolean, String, MockProducer<K, V>> producerProvider,
+			@Nullable String defaultTxId) {
+
+		this.producerProvider = producerProvider;
+		this.defaultTxId = defaultTxId;
+		this.transactional = true;
+	}
+
+	@Override
+	public boolean transactionCapable() {
+		return this.transactional;
+	}
+
+	@Override
+	public Producer<K, V> createProducer() {
+		return createProducer(this.defaultTxId);
+	}
+
+	@Override
+	public Producer<K, V> createProducer(@Nullable String txIdPrefix) {
+		return txIdPrefix == null && this.defaultTxId == null
+				? this.producerProvider.apply(false, null)
+				: this.producerProvider.apply(true, txIdPrefix == null ? this.defaultTxId : txIdPrefix);
+	}
+
+	@Override
+	public Producer<K, V> createNonTransactionalProducer() {
+		return this.producerProvider.apply(false, null);
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/mock/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes to support the use of MockConsumer and MockProducer.
+ */
+package org.springframework.kafka.mock;