> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ahmadraza.in/llms.txt
> Use this file to discover all available pages before exploring further.

# Kafka consumer PY

# 📜 **Kafka Consumer Python Script** 📜

This script demonstrates how to consume messages from a Kafka topic using the Confluent Kafka Python client. It connects to a Kafka broker, subscribes to a topic, and processes incoming messages.

## **Script Overview**

```python theme={null}
from confluent_kafka import Consumer, KafkaException, KafkaError
import sys

# Kafka broker address
bootstrap_servers = 'kafka.example.in:9092'

# Kafka topic to consume from
topic = 'web_logs_topic'  # Update this with your topic name

# Consumer configuration
conf = {
    'bootstrap.servers': bootstrap_servers,
    'group.id': 'my_consumer_group',  # Consumer group ID
    'auto.offset.reset': 'earliest'   # Start consuming from the earliest message (optional)
}

# Create Kafka Consumer
consumer = Consumer(conf)

# Subscribe to topic
consumer.subscribe([topic])

try:
    while True:
        # Poll for messages
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # End of partition, continue to next
                continue
            else:
                # Other errors
                print(msg.error())
                break

        # Print message value
        print('Received message: {}'.format(msg.value().decode('utf-8')))

except KeyboardInterrupt:
    pass

finally:
    # Close the consumer
    consumer.close()
```

## **Configuration**

### **Kafka Broker Address**

* **`bootstrap_servers`**: The address of your Kafka broker (e.g., `'kafka.example.in:9092'`).

### **Kafka Topic**

* **`topic`**: The Kafka topic you want to consume from (e.g., `'web_logs_topic'`).

### **Consumer Configuration**

* **`group.id`**: The ID for the consumer group. Consumers with the same group ID share the messages.
* **`auto.offset.reset`**: Defines where to start consuming messages if no previous offset is found. `'earliest'` starts from the beginning.

## **Handling Messages**

### **Polling Messages**

* **`consumer.poll(1.0)`**: Polls for messages with a timeout of 1 second.
* **`msg`**: The fetched message object.

### **Error Handling**

* **`msg.error()`**: Checks for errors in the message.
* **`KafkaError._PARTITION_EOF`**: Indicates end-of-partition. Continue to the next partition.
* **Other Errors**: Print the error message and stop processing.

### **Message Processing**

* **`msg.value().decode('utf-8')`**: Decodes the message value from bytes to a UTF-8 string.

## **Graceful Shutdown**

### **Keyboard Interrupt**

* **`KeyboardInterrupt`**: Handles interruption (e.g., pressing `Ctrl+C`).

### **Finally Block**

* **`consumer.close()`**: Ensures the consumer is properly closed when the script ends.

## **Running the Script**

1. **Install Dependencies**: Make sure you have `confluent-kafka` installed.
   ```bash theme={null}
   pip install confluent_kafka
   ```

2. **Run the Script**: Execute the script with Python.
   ```bash theme={null}
   python kafka_consumer.py
   ```

***

> This documentation provides a complete overview of the Kafka consumer script, including its configuration, functionality, and error handling. It helps in managing and processing Kafka messages efficiently.