diff options
Diffstat (limited to 'kafka/producer/sender.py')
| -rw-r--r-- | kafka/producer/sender.py | 9 |
1 files changed, 7 insertions, 2 deletions
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py index 72a15bb..ffc67f8 100644 --- a/kafka/producer/sender.py +++ b/kafka/producer/sender.py @@ -291,7 +291,11 @@ class Sender(threading.Thread): buf = batch.records.buffer() produce_records_by_partition[topic][partition] = buf - if self.config['api_version'] >= (0, 10): + kwargs = {} + if self.config['api_version'] >= (0, 11): + version = 3 + kwargs = dict(transactional_id=None) + elif self.config['api_version'] >= (0, 10): version = 2 elif self.config['api_version'] == (0, 9): version = 1 @@ -302,7 +306,8 @@ class Sender(threading.Thread): timeout=timeout, topics=[(topic, list(partition_info.items())) for topic, partition_info - in six.iteritems(produce_records_by_partition)] + in six.iteritems(produce_records_by_partition)], + **kwargs ) def wakeup(self): |