### Question:
KafkaProducerCallbackApplication { public void produce(final String message) { final String[] parts = message.split("-"); final String key, value; if (parts.length > 1) { key = parts[0]; value = parts[1]; } else { key = "NO-KEY"; value = parts[0]; } final ProducerRecord<String, String> producerRecord = new ProducerRecord<>(outTopic, key, value); producer.send(producerRecord, (recordMetadata, exception) -> { if (exception == null) { System.out.println("Record written to offset " + recordMetadata.offset() + " timestamp " + recordMetadata.timestamp()); } else { System.err.println("An error occurred"); exception.printStackTrace(System.err); } }); } KafkaProducerCallbackApplication(final Producer<String, String> producer,
                                    final String topic); void produce(final String message); void shutdown(); static Properties loadProperties(String fileName); static void main(String[] args);  }### Answer:
@Test public void testProduce() throws IOException { final MockProducer<String, String> mockProducer = new MockProducer<>(); final Properties props = KafkaProducerCallbackApplication.loadProperties(TEST_CONFIG_FILE); final String topic = props.getProperty("output.topic.name"); final KafkaProducerCallbackApplication producerApp = new KafkaProducerCallbackApplication(mockProducer, topic); final List<String> records = Arrays.asList("foo-bar", "bar-foo", "baz-bar", "great:weather"); records.forEach(producerApp::produce); final List<KeyValue<String, String>> expectedList = Arrays.asList(KeyValue.pair("foo", "bar"), KeyValue.pair("bar", "foo"), KeyValue.pair("baz", "bar"), KeyValue.pair("NO-KEY","great:weather")); final List<KeyValue<String, String>> actualList = mockProducer.history().stream().map(this::toKeyValue).collect(Collectors.toList()); assertThat(actualList, equalTo(expectedList)); producerApp.shutdown(); }