### Question:
KafkaProducer { public Disposable sendAndSubscribe(String topic, String key, Object value, Scheduler scheduler) { return sendOn(topic, key, value, scheduler) .subscribe(); }  Disposable sendAndSubscribe(String topic, String key, Object value, Scheduler scheduler); Disposable sendAndSubscribe(ProducerEvent producerEvent, Scheduler scheduler); Mono<SendResult<String, String>> sendOn(String topic, String key, Object value, Scheduler scheduler); Mono<SendResult<String, String>> sendOn(ProducerEvent producerEvent, Scheduler scheduler); Mono<SendResult<String, String>> send(String topic, String key, Object value); Mono<SendResult<String, String>> send(ProducerEvent producerEvent);  }### Answer:
@Test void testSendAndSubscribe() { kafkaProducer.sendAndSubscribe(SAMPLE_TOPIC, "key", "kafka value", Schedulers.elastic()); ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, SAMPLE_TOPIC); Assertions.assertEquals(record.key(), "key"); Assertions.assertEquals(record.value(), "kafka value"); }