From cab017a9aa92fa63550eb72e1571508b431ce791 Mon Sep 17 00:00:00 2001
From: Niek Sanders <niek.sanders@gmail.com>
Date: Wed, 25 Dec 2013 22:00:21 -0800
Subject: Replaced _send_upstream datetime logic with simpler time().

---
 kafka/producer.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/kafka/producer.py b/kafka/producer.py
index 7ef7896..1d47336 100644
--- a/kafka/producer.py
+++ b/kafka/producer.py
@@ -1,10 +1,10 @@
 from collections import defaultdict
-from datetime import datetime, timedelta
 from itertools import cycle
 from multiprocessing import Queue, Process
 from Queue import Empty
 import logging
 import sys
+import time
 
 from kafka.common import ProduceRequest
 from kafka.common import FailedPayloadsException
@@ -36,7 +36,7 @@ def _send_upstream(topic, queue, client, batch_time, batch_size,
     while not stop:
         timeout = batch_time
         count = batch_size
-        send_at = datetime.now() + timedelta(seconds=timeout)
+        send_at = time.time() + timeout
         msgset = defaultdict(list)
 
         # Keep fetching till we gather enough messages or a
@@ -54,7 +54,7 @@ def _send_upstream(topic, queue, client, batch_time, batch_size,
 
             # Adjust the timeout to match the remaining period
             count -= 1
-            timeout = (send_at - datetime.now()).total_seconds()
+            timeout = send_at - time.time()
             msgset[partition].append(msg)
 
         # Send collected requests upstream
-- 
cgit v1.2.1