@Test void testSendingMessagesUsingMultipleProducers() throws ExecutionException, InterruptedException { ProducerFactory<String, String> producerFactory = producerFactory(kafkaBroker); List<Producer<String, String>> testProducers = new ArrayList<>(); String testTopic = "testSendingMessagesUsingMultipleProducers"; List<Future<RecordMetadata>> results = new ArrayList<>(); for (int i = 0; i < 12; i++) { Producer<String, String> producer = producerFactory.createProducer(); results.add(send(producer, testTopic, "foo" + i)); producer.close(); testProducers.add(producer); } assertOffsets(results); cleanup(producerFactory, testProducers); }
@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();  }