ad;

moment/views/consum.py

@@ -27,10 +27,13 @@ clientid = "6"
 
 
 def kafka_consum(topic_name=None):
-    topic_name = settings.KAFKA_TOPIC_NAME if not topic_name else topic_name
-    consumser_obj = KafkaConsumer(bootstrap_servers=settings.KAFKA_BROKER_LIST, group_id="vest", client_id=clientid)
-    partition = TopicPartition(topic_name, int(clientid))
-    consumser_obj.assign([partition])
+    topic_name = settings.KAFKA_TOPIC_NAME
+    consumser_obj = KafkaConsumer(topic_name, bootstrap_servers=settings.KAFKA_BROKER_LIST, enable_auto_commit=True,
+                                  auto_commit_interval_ms=1, group_id="vest", auto_offset_reset='earliest')
+    consumser_obj.subscribe([topic_name, ])
+
+    # partition = TopicPartition(topic_name, int(clientid))
+    # consumser_obj.assign([partition])
     try:
         # redis_topic_partition_name = "vest:topic_name:" + str(topic_name)
         # topic_partition_info = redis_client.hgetall(redis_topic_partition_name)