Timestamp fix

Added pull request wintoncode#58
esall · Apr 23, 2019 · fff8a83 · fff8a83
1 parent 5867a1c
commit fff8a83
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/winton_kafka_streams/processor/_record_collector.py b/winton_kafka_streams/processor/_record_collector.py
@@ -35,6 +35,11 @@ def send(self, topic, key, value, timestamp,
 
         while not produced:
             try:
+                if isinstance(timestamp, tuple) and timestamp:
+                    timestamp = timestamp[-1]
+                if isinstance(timestamp, float):
+                    timestamp = int(timestamp)
+
                 self.producer.produce(topic, ser_value, ser_key, partition, self.on_delivery, partitioner, timestamp)
                 self.producer.poll(0)  # Ensure previous message's delivery reports are served
                 produced = True