ccloud-venv/

FROM python:3.7-slim

COPY requirements.txt /tmp/requirements.txt

RUN pip3 install -U -r /tmp/requirements.txt

COPY *.py ./

Produce messages to and consume messages from a Kafka cluster using [Confluent Python Client for Apache Kafka](https://github.com/confluentinc/confluent-kafka-python).

Fill in the commented out code in the producer and consumer files. Answers are in the `answers` branch

from confluent_kafka.avro import AvroConsumer

from confluent_kafka.avro.serializer import SerializerError

c = AvroConsumer({

'bootstrap.servers': 'localhost:32772,localhost:32773,localhost:32774',

'group.id': 'avro-consumer',

'schema.registry.url': 'http://localhost:8081',

'auto.offset.reset': 'earliest'

})

c.subscribe(['my_topic'])

while True:

try:

msg = c.poll(1.0)

except SerializerError as e:

print("Message deserialization failed for {}: {}".format(msg, e))

break

if msg is None:

continue

if msg.error():

print("AvroConsumer error: {}".format(msg.error()))

continue

print(msg.value())

c.close()

from confluent_kafka import avro

from confluent_kafka.avro import AvroProducer

value_schema_str = """

key_schema_str = """

value_schema = avro.loads(value_schema_str)

key_schema = avro.loads(key_schema_str)

value = {"name": "Value"}

key = {"name": "Key"}

def delivery_report(err, msg):

""" Called once for each message produced to indicate delivery result.

if err is not None:

print('Message delivery failed: {}'.format(err))

else:

print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

avroProducer = AvroProducer({

'bootstrap.servers': 'localhost:32772,localhost:32773,localhost:32774',

'on_delivery': delivery_report,

'schema.registry.url': 'http://localhost:8081'

}, default_key_schema=key_schema, default_value_schema=value_schema)

avroProducer.produce(topic='my_topic', value=value, key=key)

avroProducer.flush()

from confluent_kafka import Consumer

import json

if __name__ == '__main__':

# Read arguments and configurations and initialize

topic = 'temperature'

# Create Consumer instance

# '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

})

# Subscribe to topic

consumer.subscribe([topic])

# Process messages

total_count = 0

temps = []

try:

while True:

msg = consumer.poll(1.0)

if msg is None:

# No message available within timeout.

# Initial message consumption may take up to

# `session.timeout.ms` for the consumer group to

# rebalance and start consuming

print("Waiting for message or event/error in poll()")

continue

elif msg.error():

print('error: {}'.format(msg.error()))

else:

# Exercise: Read and print the key and value

pass

except KeyboardInterrupt:

pass

finally:

# Leave group and commit final offsets

consumer.close()

from confluent_kafka import Producer, KafkaError

import json

import random

from confluent_kafka.admin import AdminClient

from confluent_kafka.cimpl import NewTopic

if __name__ == '__main__':

topic = 'temperature'

admin_client = AdminClient({'bootstrap.servers': 'localhost:32768,localhost:32769,localhost:32770'})

admin_client.create_topics([NewTopic(topic, num_partitions=3, replication_factor=1)])

# Create Producer instance

producer = Producer({

# Exercise: Add your producer configuration here

})

delivered_records = 0

# Optional per-message on_delivery handler (triggered by poll() or flush())

# when a message has been successfully delivered or

# permanently failed delivery (after retries).

def acked(err, msg):

global delivered_records

"""Delivery report handler called on

if err is not None:

print("Failed to deliver message: {}".format(err))

else:

delivered_records += 1

print("Produced record to topic {} partition [{}] @ offset {}"

.format(msg.topic(), msg.partition(), msg.offset()))

cities = ['London', 'New York', 'Madrid', 'Paris']

temps = [10, 20, 30]

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

# p.poll() serves delivery reports (on_delivery)

# from previous produce() calls.

producer.poll(0)

producer.flush()

print("{} messages were produced to topic {}!".format(delivered_records, topic))

confluent-kafka[avro,json,protobuf]>=1.4.2