answers

githubmo · githubmo · commit 5718d4853700 · 2020-08-13T07:22:13.000+01:00
diff --git a/consumer.py b/consumer.py
@@ -34,8 +34,9 @@
     # 'auto.offset.reset=earliest' to start reading from the beginning of the
     #   topic if no committed offsets exist
     consumer = Consumer({
-        # Exercise: Add the consumer setting
-        # Make sure you add a consumer group and choose to reset offset to earliest
+        'bootstrap.servers': 'localhost:32768,localhost:32769,localhost:32770',
+        'group.id': 'python_example_group_1',
+        'auto.offset.reset': 'earliest',
     })
 
     # Subscribe to topic
@@ -57,10 +58,14 @@
             elif msg.error():
                 print('error: {}'.format(msg.error()))
             else:
-
-                # Exercise: Read and print the key and value
-                pass
-
+                # Check for Kafka message
+                record_key = msg.key()
+                record_value = msg.value()
+                print(record_value)
+                data = json.loads(record_value)
+                print("Consumed record with key {} and value {}, \
+                      and updated total count to {}"
+                      .format(record_key, record_value, 0))
     except KeyboardInterrupt:
         pass
     finally:
diff --git a/producer.py b/producer.py
@@ -39,7 +39,7 @@
 
     # Create Producer instance
     producer = Producer({
-        # Exercise: Add your producer configuration here
+        'bootstrap.servers': 'localhost:32768,localhost:32769,localhost:32770'
     })
 
     delivered_records = 0
@@ -65,10 +65,11 @@ def acked(err, msg):
     for n in range(10):
         city = random.choice(cities)
         temp = random.choice(temps)
-
-        # Exercise: Create a Json with the city and temp
-        # Send a message to kafka with the city as key and the json as value
-
+        j = json.dumps({'city': city, 'temp': temp})
+        record_key = city
+        record_value = json.dumps(j)
+        print("Producing record: {}\t{}".format(record_key, record_value))
+        producer.produce(topic, key=record_key, value=record_value, on_delivery=acked)
         # p.poll() serves delivery reports (on_delivery)
         # from previous produce() calls.
         producer.poll(0)
