Fixed startnode button showing for startnode

functions/extensions/kafka/kafka_local.py

@@ -11,44 +11,77 @@ shuffle_workflow = os.getenv("SHUFFLE_WORKFLOW")
     
 headers = {"Authorization": "Bearer %s" % shuffle_apikey}
 topic = "workflow_%s" % shuffle_workflow
+group = "testing"
 server = "localhost:9092"
 def produce():
-    print("Starting producer")
-    producer = KafkaProducer(
-        bootstrap_servers=[server],
-        value_serializer=lambda x:
-        json.dumps(x).encode('utf-8')
-    )
-    
-    print("Adding data!")
-    for e in range(15):
-        data = {"some": e, "data": "luuuuul"}
+    try:
+        #group_id=group,
+        #value_serializer=lambda x: json.dumps(x).encode('utf-8')
+        producer = KafkaProducer(
+            bootstrap_servers=[server],
+        )
+    except kafka.errors.NoBrokersAvailable as e:
+        print(f"Error with producer: {e}")
+        sleep(5)
+        produce()
+        return
+    except ValueError as e:
+        print(f"ValuError with producer: {e}")
+        sleep(5)
+        produce()
+        return
+        
+        
+    for i in range(15):
+        producer.send(topic, json.dumps({"some": i, "data": "luuuuul"}))
 
-        try:
-            ret = producer.send(topic, value=data)
-            print(ret.get())
-        except kafka.errors.KafkaTimeoutError as e:
-            print("Kafka error: %s" % e)
-            continue
+    producer.flush()
+    producer.close()
+    exit()
+
+        #print(f"Adding data {data}")
+
+        #try:
+        #    ret = producer.send(topic, value=data)
+        #    print(ret.get())
+        #    producer.flush()
+
+        #    #future = producer.send('foobar', b'another_message')
+        #    #result = future.get(timeout=60)
+        #    #print(result)
+
+        #except kafka.errors.KafkaTimeoutError as e:
+        #    print("Kafka error: %s" % e)
+        #    continue
 
 def consume():
     print("Starting consumer")
+    #group_id=group,
+    #auto_offset_reset="earliest",
+    #enable_auto_commit=True,
     consumer = KafkaConsumer(
         topic,
         bootstrap_servers=[server],
-        auto_offset_reset="earliest",
-        enable_auto_commit=True,
-        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
+        auto_offset_reset='earliest',
+        max_poll_records=2,
     )
+    #value_deserializer=lambda x: json.loads(x.decode('utf-8'))
 
-    print("Getting data")
+    consumer.poll()
+    #consumer.seek_to_beginning()
+
+    print(f"Getting data from topic {topic}")
     for message in consumer:
-        message = message.value
+        #message = message.value
+        #message = message
         print("MSG: ", message)
         ret = requests.post("%s/api/v1/%s/execute" % shuffle_url, headers=headers, data=message)
         print(ret.status_code)
         print(ret.text)
 
 if __name__ == "__main__":
+    if topic == "workflow_None":
+        topic = "testing"
+    print("Starting producer on %s for topic %s" % (server, topic))
     produce()
     #consume()