### Question:
MockKafka implements Kafka<K, V> { public MockKafka<K, V> withRecordMapper(BiFunction<ProducerRecord<K, V>, RecordMetadata, ConsumerRecord<K, V>> recordMapper) { this.recordMapper = recordMapper; return this; } MockKafka();  MockKafka(int maxPartitions, int maxHistory); MockKafka<K, V> withSendCallbackExceptionGenerator(ExceptionGenerator<ProducerRecord<K, V>, Exception> sendCallbackExceptionGenerator); MockKafka<K, V> withSendRuntimeExceptionGenerator(ExceptionGenerator<ProducerRecord<K, V>, RuntimeException> sendRuntimeExceptionGenerator); MockKafka<K, V> withCommitExceptionGenerator(ExceptionGenerator<Map<TopicPartition, OffsetAndMetadata>, Exception> commitExceptionGenerator); MockKafka<K, V> withAdminClientFactory(Supplier<AdminClient> adminClientFactory); MockKafka<K, V> withRecordMapper(BiFunction<ProducerRecord<K, V>, RecordMetadata, ConsumerRecord<K, V>> recordMapper); @Override void describeProducer(LogLine logLine, Properties defaults, Properties overrides); @Override FallibleMockProducer<K, V> getProducer(Properties overrides); @Override FallibleMockProducer<K, V> getProducer(Properties defaults, Properties overrides); List<ConsumerRecord<K, V>> getBacklog(); @Override void describeConsumer(LogLine logLine, Properties defaults, Properties overrides); @Override FallibleMockConsumer<K, V> getConsumer(Properties overrides); @Override FallibleMockConsumer<K, V> getConsumer(Properties defaults, Properties overrides); @Override AdminClient getAdminClient(); @Override String toString();  }### Answer:
@Test public void testWithRecordMapper() { ConsumerRecord<Object, Object> expectedRecord = new ConsumerRecord<>("topic", 0, 0, "key", "value"); Kafka<Object, Object> mockKafka = new MockKafka<>() .withRecordMapper((objectObjectProducerRecord, recordMetadata) -> expectedRecord); final SerdeProps props = new SerdeProps(SerdePair.STRING); final Producer<Object, Object> producer = mockKafka.getProducer(props.producer()); Consumer<Object, Object> consumer = mockKafka.getConsumer(new Properties()); consumer.subscribe(Collections.singleton("topic")); producer.send(new ProducerRecord<>("topic", 0, "key", "value")); wait.until(() -> { ConsumerRecords<Object, Object> consumerRecords = consumer.poll(Duration.ofMillis(1)); assertEquals(1, consumerRecords.count()); assertSame(expectedRecord, consumerRecords.iterator().next()); }); }