### Question:
MockKafka implements Kafka<K, V> { ConsumerRecord<K, V> defaultRecordMapping(ProducerRecord<K, V> record, RecordMetadata metadata) { final int partition = record.partition() != null ? record.partition() : metadata.partition(); return new ConsumerRecord<>(record.topic(), partition, metadata.offset(), metadata.timestamp(), TimestampType.NO_TIMESTAMP_TYPE, -1L, -1, -1, record.key(), record.value(), record.headers()); } 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 testDefaultRecordMapping() { final MockKafka<Object, Object> mockKafka = new MockKafka<>(); final RecordHeaders recordHeaders = new RecordHeaders(Collections.singleton( new RecordHeader("headerKey", "headerValue".getBytes(StandardCharsets.UTF_8)))); final RecordMetadata recordMetadata = new RecordMetadata(new TopicPartition("topic", 0), 0, 0, 0, -1L, -1, -1); final ProducerRecord<Object, Object> producerRecord = new ProducerRecord<>("topic", 0, "key", "value", recordHeaders); final ConsumerRecord<Object, Object> consumerRecord = mockKafka.defaultRecordMapping(producerRecord, recordMetadata); assertEquals(producerRecord.topic(), consumerRecord.topic()); assertEquals(producerRecord.partition().intValue(), consumerRecord.partition()); assertEquals(producerRecord.key(), consumerRecord.key()); assertEquals(producerRecord.value(), consumerRecord.value()); assertEquals(producerRecord.headers(), consumerRecord.headers()); }