produce.java

  public <K, V> void produce(String producerName, int messageCount, Serializer<K> keySerializer, Serializer<V> valueSerializer, Runnable completionCallback, Supplier<ProducerRecord<K, V>> messageSupplier) {

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", kafkaCluster.bootstrapServers());
        props.setProperty("acks", Integer.toString(1));
        Thread t = new Thread(() -> {
            LOGGER.info("Starting producer {} to write {} messages", producerName, messageCount);
            try {
                KafkaProducer<K, V> producer = new KafkaProducer<>(props, keySerializer, valueSerializer);
                Throwable cause = null;

                try {
                    for (int i = 0; i != messageCount; ++i) {
                        ProducerRecord<K, V> record = messageSupplier.get();
                        producer.send(record);
                        producer.flush();
                        LOGGER.info("Producer {}: sent message {}", producerName, record);
                    }
                } catch (Throwable e) {
                    cause = e;
                    throw e;
                } finally {
                    if (cause != null) {
                        try {
                            producer.close();
                        } catch (Throwable c) {
                            cause.addSuppressed(c);
                        }
                    } else {
                        producer.close();
                    }
                }
            } finally {
                if (completionCallback != null) {
                    completionCallback.run();
                }
                LOGGER.debug("Stopping producer {}", producerName);
            }
        });
        t.setName(producerName + "-thread");
        t.start();
    }