### Question:
KafkaProducer { public Mono<SendResult<String, String>> send(String topic, String key, Object value) { return Mono.fromCallable(() -> ProducerEvent.builder().topic(topic).key(key).value(value).build()) .flatMap(this::send); }  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 testSendSuccess() { kafkaProducer.send(SAMPLE_TOPIC, "key", "kafka value").subscribe(); ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, SAMPLE_TOPIC); Assertions.assertEquals(record.key(), "key"); Assertions.assertEquals(record.value(), "kafka value"); }
@Test void testSendLazy() { Assertions.assertThrows(IllegalStateException.class, () -> { kafkaProducer.send(SAMPLE_TOPIC, "key", "kafka value"); ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, SAMPLE_TOPIC, 5_000L); }); }