#!/usr/bin/env python

#

# Copyright 2020 Confluent Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

#

# =============================================================================

#

# Consume messages from Confluent Cloud

# Using Confluent Python Client for Apache Kafka

#

# =============================================================================

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()