#!/usr/local/bin/python

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import ssl
import os
import json
import getopt
import sys

import logging
import sys
logger = logging.getLogger('kafka')
logger.addHandler(logging.StreamHandler(sys.stdout))
logger.setLevel(logging.INFO)

opts = {}

if len(sys.argv) != 2:
    print('Usage: kafka_receive <topic-name>')
    sys.exit(-1)

opts['topic'] = sys.argv[1]

print('Attempting to connect to topic: ' + opts['topic'])

if os.environ.get('VCAP_SERVICES'):
    # Running in Bluemix
    print('Running in Bluemix mode.')
    vcap_services = json.loads(os.environ.get('VCAP_SERVICES'))
    for vcap_service in vcap_services:
        if vcap_service.startswith('messagehub'):
            messagehub_service = vcap_services[vcap_service][0]
            opts['brokers'] = ','.join(messagehub_service['credentials']['kafka_brokers_sasl'])
            opts['api_key'] = messagehub_service['credentials']['api_key']
            opts['username'] = messagehub_service['credentials']['user']
            opts['password'] = messagehub_service['credentials']['password']
            opts['rest_endpoint'] = messagehub_service['credentials']['kafka_admin_url']
else:
    print('ERROR: no VCAP_SERVICES found in environment')
    sys.exit(-1)

if opts == {}:
    print('ERROR: no messagehub bound to application')
    sys.exit(-1)


sasl_mechanism = 'PLAIN'
security_protocol = 'SASL_SSL'

# Create a new context using system defaults, disable all but TLS1.2
context = ssl.create_default_context()
context.options &= ssl.OP_NO_TLSv1
context.options &= ssl.OP_NO_TLSv1_1

print ('connecting')

# always use a new groupid to get all the data
import uuid;
group_id = uuid.uuid4()

consumer = KafkaConsumer(
                         bootstrap_servers = opts['brokers'],
                         sasl_plain_username = opts['username'],
                         sasl_plain_password = opts['password'],
                         security_protocol = security_protocol,
                         ssl_context = context,
                         sasl_mechanism = sasl_mechanism,
                         api_version = (0,10),
                         auto_offset_reset = 'latest',
                         group_id = group_id)

consumer.subscribe(opts['topic'])

print ('connected')

for msg in consumer:
    print(msg)


@@ -341,11 +341,60 @@
  != 
-2:%0A
+3 or sys.argv%5B2%5D not in ('latest', 'earliest'):%0A    
     
@@ -433,19 +433,39 @@
 ic-name%3E
-')%0A
+ latest%7Cearliest')%0A    
     sys.
@@ -501,16 +501,56 @@
 .argv%5B1%5D
+%0Aopts%5B'auto.offset.reset'%5D = sys.argv%5B2%5D
 %0A%0Aprint(
@@ -1806,16 +1806,63 @@
 he data%0A
+if opts%5B'auto.offset.reset'%5D == 'earliest':%0A   
 import u
@@ -1866,16 +1866,19 @@
 t uuid;%0A
+   
 group_id
@@ -1892,16 +1892,62 @@
 .uuid4()
+%0Aelse:%0A   group_id = 'console-consumer-script'
 %0A%0Aconsum
@@ -2420,24 +2420,41 @@
 reset = 
-'lat
+opts%5B'auto.offset.r
 es
+e
 t'
+%5D
 ,%0A      
@@ -2494,16 +2494,219 @@
 up_id)%0A%0A
+if opts%5B'topic'%5D not in consumer.topics():%0A    print(%22%5Cn** ERROR: Topic '%7B%7D' not found - have you spelt it correctly? Available topics: %7B%7D **%22.format(opts%5B'topic'%5D, consumer.topics()))%0A    sys.exit(-1)%0A%0A
 consumer

Check topic exists before consuming