MessageProducer { public void send(ProducerRecord<byte[], byte[]> record) { Producer<byte[], byte[]> producer = this.producer; if (producer == null) { synchronized (this) { if (this.producer == null) tryCreateProducer(); if (this.producer == null) throw new KafkaException("kafka uri is not resolvable, uri=" + uri); producer = this.producer; } } producer.send(record, new KafkaCallback(record)); } MessageProducer(KafkaURI uri, String name); void send(ProducerRecord<byte[], byte[]> record); void tryCreateProducer(); void close(long timeoutInMs); final ProducerMetrics producerMetrics; }
@Test void send() { var record = new ProducerRecord<byte[], byte[]>("topic", Strings.bytes("{}")); when(uri.resolveURI()).thenReturn(Boolean.FALSE); assertThatThrownBy(() -> producer.send(record)) .isInstanceOf(KafkaException.class) .hasMessageContaining("kafka uri is not resolvable"); }