### Question:
KafkaProducer { public Mono<SendResult<String, String>> sendOn(String topic, String key, Object value, Scheduler scheduler) { return send(topic, key, value) .subscribeOn(scheduler); }  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 testSendOn() { kafkaProducer.sendOn(SAMPLE_TOPIC, "key", "kafka value", Schedulers.elastic()).subscribe(); ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, SAMPLE_TOPIC); Assertions.assertEquals(record.key(), "key"); Assertions.assertEquals(record.value(), "kafka value"); }
@Test void testSendOnLazy() { Assertions.assertThrows(IllegalStateException.class, () -> { kafkaProducer.sendOn(SAMPLE_TOPIC, "key", "kafka value", Schedulers.elastic()); ConsumerRecord<String, String> record = KafkaTestUtils.getSingleRecord(consumer, SAMPLE_TOPIC, 5_000L); }); }