consumer.py

# consumer.py

from kafka import KafkaConsumer
import json
import requests

consumer = KafkaConsumer(
    'coordinates',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    group_id='coordinate-group',
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))
)

print("Waiting for messages...")
counter = 0
payload = {"x" : [],"y":[]}
for message in consumer:
    print(counter)
    counter +=1
    coords = message.value
    payload["x"].append(coords['x'])
    payload['y'].append(coords['y'])
    if(counter == 100):
        if "id" not in payload.keys():
            payload["id"]=message.key.decode('utf-8')
        response=requests.post("http://localhost:8000/ingestdata",json=payload)
        if response.status_code == 200:
            counter = 0
            payload = {"x" : [],"y":[]}
    #print(payload)
    #
    #print(f"Received: x={coords['x']}, y={coords['y']}")