@Override
        public void run() {
            try {
                Map<String, NewTopic> newTopics = new HashMap<>();
                if (spec.activeTopic().materialize().size() != 1) {
                    throw new RuntimeException("Can only run against 1 topic.");
                }
                List<TopicPartition> active = new ArrayList<>();
                for (Map.Entry<String, PartitionsSpec> entry :
                        spec.activeTopic().materialize().entrySet()) {
                    String topicName = entry.getKey();
                    PartitionsSpec partSpec = entry.getValue();
                    newTopics.put(topicName, partSpec.newTopic(topicName));
                    for (Integer partitionNumber : partSpec.partitionNumbers()) {
                        active.add(new TopicPartition(topicName, partitionNumber));
                    }
                }
                status.update(new TextNode("Creating " + newTopics.keySet().size() + " topic(s)"));
                WorkerUtils.createTopics(log, spec.bootstrapServers(), spec.commonClientConf(),
                                         spec.adminClientConf(), newTopics, false);
                status.update(new TextNode("Created " + newTopics.keySet().size() + " topic(s)"));
                executor.submit(new SendRecords(active.get(0).topic(), spec.activePartition()));
            } catch (Throwable e) {
                WorkerUtils.abort(log, "Prepare", e, doneFuture);
            }
        }
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