From 9e45546b72c91805476fca417020326830fc16f2 Mon Sep 17 00:00:00 2001
From: Vincent Janelle <vjanelle@users.noreply.github.com>
Date: Mon, 30 Apr 2018 11:10:49 -0700
Subject: [PATCH] Move message timer around (#9)

In batch writes, this would produce histogram timers in seconds, since
the counter would to wait for all several thousand messages to be sent.
---
 src/main/java/com/salesforce/WriteTopic.java | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/main/java/com/salesforce/WriteTopic.java b/src/main/java/com/salesforce/WriteTopic.java
index 294a948..169d372 100644
--- a/src/main/java/com/salesforce/WriteTopic.java
+++ b/src/main/java/com/salesforce/WriteTopic.java
@@ -83,13 +83,13 @@ public Exception call() {
 
 
             while (keepProducing) {
-                produceMessageTimeSecs.time(() -> {
-                    // TODO: Get this from properties
-                    for (int i = 0; i < numMessagesToSendPerBatch; i++) {
-                        kafkaProducer.send(new ProducerRecord<>(topicName, topicId, i));
-                        log.debug("{}: Produced message {}", formatter.format(new Date()), topicId);
-                    }
-                });
+                // TODO: Get this from properties
+                for (int i = 0; i < numMessagesToSendPerBatch; i++) {
+                    Histogram.Timer requestTimer = produceMessageTimeSecs.startTimer();
+                    kafkaProducer.send(new ProducerRecord<>(topicName, topicId, i));
+                    requestTimer.observeDuration();
+                    log.debug("{}: Produced message {}", formatter.format(new Date()), topicId);
+                }
                 threadsAwaitingMessageProduce.dec();
                 Thread.sleep(readWriteInterval);
                 threadsAwaitingMessageProduce.inc();