@Test void testUsingCallbackWhilePublishingMessages() throws ExecutionException, InterruptedException { Callback cb = mock(Callback.class); ProducerFactory<String, String> pf = producerFactory(kafkaBroker); Producer<String, String> producer = pf.createProducer(); producer.send(new ProducerRecord<>("testUsingCallbackWhilePublishingMessages", "callback"), cb).get(); producer.flush(); verify(cb, only()).onCompletion(any(RecordMetadata.class), any()); cleanup(pf, producer); }
@Override public Producer<K, V> createProducer() { if (confirmationMode.isTransactional()) { return createTransactionalProducer(); } if (this.nonTransactionalProducer == null) { synchronized (this) { if (this.nonTransactionalProducer == null) { this.nonTransactionalProducer = new ShareableProducer<>(createKafkaProducer(configuration)); } } } return this.nonTransactionalProducer; }
DefaultProducerFactory implements ProducerFactory<K, V> { @Override public Producer<K, V> createProducer() { if (confirmationMode.isTransactional()) { return createTransactionalProducer(); } if (this.nonTransactionalProducer == null) { synchronized (this) { if (this.nonTransactionalProducer == null) { this.nonTransactionalProducer = new ShareableProducer<>(createKafkaProducer(configuration)); } } } return this.nonTransactionalProducer; } }
DefaultProducerFactory implements ProducerFactory<K, V> { @Override public Producer<K, V> createProducer() { if (confirmationMode.isTransactional()) { return createTransactionalProducer(); } if (this.nonTransactionalProducer == null) { synchronized (this) { if (this.nonTransactionalProducer == null) { this.nonTransactionalProducer = new ShareableProducer<>(createKafkaProducer(configuration)); } } } return this.nonTransactionalProducer; } @SuppressWarnings("WeakerAccess") protected  DefaultProducerFactory(Builder<K, V> builder); }
DefaultProducerFactory implements ProducerFactory<K, V> { @Override public Producer<K, V> createProducer() { if (confirmationMode.isTransactional()) { return createTransactionalProducer(); } if (this.nonTransactionalProducer == null) { synchronized (this) { if (this.nonTransactionalProducer == null) { this.nonTransactionalProducer = new ShareableProducer<>(createKafkaProducer(configuration)); } } } return this.nonTransactionalProducer; } @SuppressWarnings("WeakerAccess") protected  DefaultProducerFactory(Builder<K, V> builder); static Builder<K, V> builder(); @Override Producer<K, V> createProducer(); @Override ConfirmationMode confirmationMode(); Map<String, Object> configurationProperties(); String transactionIdPrefix(); @Override void shutDown(); }
DefaultProducerFactory implements ProducerFactory<K, V> { @Override public Producer<K, V> createProducer() { if (confirmationMode.isTransactional()) { return createTransactionalProducer(); } if (this.nonTransactionalProducer == null) { synchronized (this) { if (this.nonTransactionalProducer == null) { this.nonTransactionalProducer = new ShareableProducer<>(createKafkaProducer(configuration)); } } } return this.nonTransactionalProducer; } @SuppressWarnings("WeakerAccess") protected  DefaultProducerFactory(Builder<K, V> builder); static Builder<K, V> builder(); @Override Producer<K, V> createProducer(); @Override ConfirmationMode confirmationMode(); Map<String, Object> configurationProperties(); String transactionIdPrefix(); @Override void shutDown();  }