knativekafka 0.2.2

A wrapper for kafka-python used in knative

knativekafka

knativekafka is a wrapper around kafka-python package. This package is easy to use and can be used to test the Kafka Producer in kafka-python package.

Some important items to note.

• This package includes the basic testing only as of now.

• Polling, parallelism etc. has not been included yet.

Installation Instructions

The package is availble in PyPI which is an official repository for python package. pip is a package management system and is used for installing Python packages from Python Package Index (also known as PyPI). It is the most common way to install Python packages. Install the package in your environment from the terminal using the below pip command:

pip3 install knativekafka

Pre-requisites

• Python 3.x

• pip3

• Set the environment variable - KAFKA_BOOTSTRAP_SERVERS

• Have the kafka server installed and configured with a topic

How to Use?

• Import the Kafka Producer into your python code using the below import.

from knativekafka.knativekafkaproducer import KNativeKafkaProducer

• To import the Kafka Consumer use the below import. from knativekafka.knativekafkaconsumer import KNativeKafkaConsumer

Usage

• Only few features are implemented in the package as of now.

• Additional features to be included later.

Test Producer

To test the producer, perform the below steps.

• Instantiate the KNativeKafkaProducer by passing the kafka topic.

• Bootstrap Servers -- The init() method checks whether the os.environ() contains the KAFKA_BOOTSTRAP_SERVERS key.

• If so, it fetches the value from the environment. If not, it sets the default value as localhost:9092.

• Topic -- The topic is passed as parameter. Since, topic is a required field,if topic is an empty string, then, topic is retrieved from the environment variable. If topic is not passed and environment variable doesn't contain KAFKA_TOPIC, ValueError is thrown. To send data, use the send_binary_data() by passing the message as parameter. The send_binary_data() takes the message in binary format.

• TLS Enabled -- If environment variable - KAFKA_NET_TLS_ENABLE is found and has value True, then, TLS is enabled. Get the tls information such as ssl_cafile, ssl_certfile and ssl_keyfile from the corresponding environment variable. The environment variables are KAFKA_NET_TLS_CA_CERT, KAFKA_NET_TLS_CERT and KAFKA_NET_TLS_KEY.

• TLS Disabled -- If either environment variable - KAFKA_NET_TLS_ENABLE is not found or has value False, then, TLS is disabled. Set the value for the tls fields such as ssl_cafile, ssl_certfile and ssl_keyfile to None.

    topic = "python-topic"
    kafka_producer = KNativeKafkaProducer(topic=topic)
    value = input('value:')
    value = bytes(value, encoding='utf8')            
    kafka_producer.send_binary_data(value)    
    print("Successfully sent data....") 

Test Consumer

To test the consumer, perform the below steps.

• Instantiate the KNativeKafkaConsumer by passing the topic and then, call getMessage(). The getMessage() prints the message values.

• Bootstrap Servers -- The init() method checks whether the os.environ() contains the KAFKA_BOOTSTRAP_SERVERS key.

• If so, it fetches the value from the environment. If not, it sets the default value as localhost:9092.

• Topic -- The topic is passed as parameter. Since, topic is a required field,if topic is an empty string, then, topic is retrieved from the environment variable. If topic is not passed and environment variable doesn't contain KAFKA_TOPIC, ValueError is thrown. To send data, use the send_binary_data() by passing the message as parameter. The send_binary_data() takes the message in binary format.

• TLS Enabled -- If environment variable - KAFKA_NET_TLS_ENABLE is found and has value True, then, TLS is enabled. Get the tls information such as ssl_cafile, ssl_certfile and ssl_keyfile from the corresponding environment variable. The environment variables are KAFKA_NET_TLS_CA_CERT, KAFKA_NET_TLS_CERT and KAFKA_NET_TLS_KEY.

• TLS Disabled -- If either environment variable - KAFKA_NET_TLS_ENABLE is not found or has value False, then, TLS is disabled. Set the value for the tls fields such as ssl_cafile, ssl_certfile and ssl_keyfile to None.

topic = "python-topic"
kafka_consumer =  KNativeKafkaConsumer(topics=topic)
msg = kafka_consumer.getMessage()    
print(msg)

References

• knativekafka in test.pypi.org

• knativekafka in pypi.org

License

Apache License