@Override
        public Void call() throws Exception {
            long startTimeMs = Time.SYSTEM.milliseconds();
            try {
                try {
                    long sentMessages = 0;
                    while (true) {
                        sendMessage();
                        sentMessages++;
                    }
                } catch (Exception e) {
                    throw e;
                } finally {
                    if (sendFuture != null) {
                        try {
                            sendFuture.get();
                        } catch (Exception e) {
                            log.error("Exception on final future", e);
                        }
                    }
                    producer.close();
                }
            } catch (Exception e) {
                WorkerUtils.abort(log, "SendRecords", e, doneFuture);
            } finally {
                statusUpdaterFuture.cancel(false);
                StatusData statusData = new StatusUpdater(histogram).update();
                long curTimeMs = Time.SYSTEM.milliseconds();
                log.info("Sent {} total record(s) in {} ms.  status: {}",
                    histogram.summarize().numSamples(), curTimeMs - startTimeMs, statusData);
            }
            doneFuture.complete("");
            return null;
        }
This workload allows for customized and even variable configurations in terms of messages per second, message size,
batch size, key size, and even the ability to target a specific partition out of a topic.

See `ConfigurableProducerSpec` for a more detailed description.
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConfigurableProducerWorker.java