Update kafka-python to 2.0.2 to add support for python 3.8

CHANGES.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 1.1.0
+
+* Update to kafka-python 2.0.2 to support python 3.8. #16 by @murthysrd
+
 ## 1.0.0
 
 * Drop Python 2.7 support

actions/produce.py

@@ -1,13 +1,16 @@
+"""
+Kafka Producer Action for stackstorm
+"""
 from st2common.runners.base_action import Action
-from kafka import SimpleProducer, KafkaClient
-from kafka.util import kafka_bytestring
+from kafka import KafkaProducer, KafkaConsumer
 
 
 class ProduceMessageAction(Action):
     """
     Action to send messages to Apache Kafka system.
     """
-    DEFAULT_CLIENT_ID = 'st2-kafka-producer'
+
+    DEFAULT_CLIENT_ID = "st2-kafka-producer"
 
     def run(self, topic, message, hosts=None):
         """
@@ -28,18 +31,28 @@ def run(self, topic, message, hosts=None):
 
         if hosts:
             _hosts = hosts
-        elif self.config.get('hosts', None):
-            _hosts = self.config['hosts']
+        elif self.config.get("hosts", None):
+            _hosts = self.config["hosts"]
         else:
             raise ValueError("Need to define 'hosts' in either action or in config")
 
         # set default for empty value
-        _client_id = self.config.get('client_id') or self.DEFAULT_CLIENT_ID
-
-        client = KafkaClient(_hosts, client_id=_client_id)
-        client.ensure_topic_exists(topic)
-        producer = SimpleProducer(client)
-        result = producer.send_messages(topic, kafka_bytestring(message))
-
-        if result[0]:
-            return result[0]._asdict()
+        _client_id = self.config.get("client_id") or self.DEFAULT_CLIENT_ID
+
+        consumer = KafkaConsumer(
+            bootstrap_servers=_hosts.split(","), client_id=_client_id
+        )
+        try:
+            assert topic in consumer.topics()
+        except AssertionError as exc:
+            raise AssertionError(f"{topic} does not exist.") from exc
+
+        producer = KafkaProducer(
+            bootstrap_servers=_hosts.split(","),
+            client_id=_client_id,
+            value_serializer=lambda m: m.encode("utf-8"),
+            max_request_size=10485760,
+        )
+        future = producer.send(topic, message)
+        record_metadata = future.get(timeout=10)  # TODO: Make this timeout an input param
+        return record_metadata._asdict()

pack.yaml

@@ -10,7 +10,7 @@ keywords:
   - base64
   - stackdriver
   - google cloud
-version: 1.0.0
+version: 1.1.0
 author: StackStorm
 email: [email protected]
 Contributors:

requirements.txt

@@ -1 +1 @@
-kafka-python>=0.9.4,<1.0
+kafka-python==2.0.2