@Override
        public Void call() throws Exception {
            long startTimeMs = Time.SYSTEM.milliseconds();
            try {
                try {
                    if (enableTransactions)
                        producer.initTransactions();

                    long sentMessages = 0;
                    while (sentMessages < spec.maxMessages()) {
                        if (enableTransactions) {
                            boolean tookAction = takeTransactionAction();
                            if (tookAction)
                                continue;
                        }
                        sendMessage();
                        sentMessages++;
                    }
                    if (enableTransactions)
                        takeTransactionAction(); // give the transactionGenerator a chance to commit if configured evenly
                } catch (Exception e) {
                    if (enableTransactions)
                        producer.abortTransaction();
                    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, transactionsCommitted).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;
        }
A subclass of Throttle which flushes the Producer right before the throttle injects a delay.
This avoids including throttling latency in latency measurements.
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java