fixup reset_producer_id is_transactional() check

Author: dpkp

kafka/producer/transaction_manager.py

@@ -47,6 +47,9 @@ def is_valid(self):
     def match(self, batch):
         return self.producer_id == batch.producer_id and self.epoch == batch.producer_epoch
 
+    def __eq__(self, other):
+        return isinstance(other, ProducerIdAndEpoch) and self.producer_id == other.producer_id and self.epoch == other.epoch
+
     def __str__(self):
         return "ProducerIdAndEpoch(producer_id={}, epoch={})".format(self.producer_id, self.epoch)
 
@@ -304,7 +307,7 @@ def reset_producer_id(self):
         it's best to return the produce error to the user and let them abort the transaction and close the producer explicitly.
         """
         with self._lock:
-            if self.is_transactional:
+            if self.is_transactional():
                 raise Errors.IllegalStateError( 
                     "Cannot reset producer state for a transactional producer."
                     " You must either abort the ongoing transaction or"

test/test_producer.py

@@ -7,6 +7,9 @@
 import pytest
 
 from kafka import KafkaProducer
+from kafka.cluster import ClusterMetadata
+from kafka.producer.transaction_manager import TransactionManager, ProducerIdAndEpoch
+
 
 @pytest.mark.skipif(platform.python_implementation() != 'CPython',
                     reason='Test relies on CPython-specific gc policies')
@@ -20,4 +23,17 @@ def test_kafka_producer_gc_cleanup():
     assert threading.active_count() == threads
 
 
+def test_idempotent_producer_reset_producer_id():
+    transaction_manager = TransactionManager(
+        transactional_id=None,
+        transaction_timeout_ms=1000,
+        retry_backoff_ms=100,
+        api_version=(0, 11),
+        metadata=ClusterMetadata(),
+    )
 
+    test_producer_id_and_epoch = ProducerIdAndEpoch(123, 456)
+    transaction_manager.set_producer_id_and_epoch(test_producer_id_and_epoch)
+    assert transaction_manager.producer_id_and_epoch == test_producer_id_and_epoch
+    transaction_manager.reset_producer_id()
+    assert transaction_manager.producer_id_and_epoch == ProducerIdAndEpoch(-1, -1)