Notebook - Asynchronous Producer¶

# Topic topic_example_v1 has been created

from confluent_kafka.admin import AdminClient, NewTopic

# def load_config():
#     """Load Kafka configuration."""
#     return {
#         'bootstrap.servers': '{server}',
#         'security.protocol': '{}',
#         'sasl.mechanisms': '{}',
#         'sasl.username': '{api key}',
#         'sasl.password': '{api password}'
#     }
    
## Recommended way of loading secrets from .env file
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def load_config():
    """Load Kafka configuration."""
    return {
        'bootstrap.servers': os.getenv('BOOTSTRAP_SERVERS'),
        'security.protocol': os.getenv('SECURITY_PROTOCOL'),
        'sasl.mechanisms': os.getenv('SASL_MECHANISMS'),
        'sasl.username': os.getenv('SASL_USERNAME'),
        'sasl.password': os.getenv('SASL_PASSWORD')
    }

# in python 
config = load_config()

# Create callback function

# Producer callback
def delivery_report(err, msg):
    """Callback to report the result of the produce operation."""
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

# Define message to generate

import random
def produce_messages(producer, topic_name, num_messages=10):
    """Produce messages with the given format."""
    names = ["Alice", "Bob", "Charlie"]

    for i in range(num_messages):
        name = random.choice(names)
        lat = random.uniform(-90, 90)
        long = random.uniform(-180, 180)
        message_key = f"rider-{name}-{i}"  # Just as an example. Choose a meaningful key if needed.
        message_value = f"rider {name} requests a car at ({lat:.2f}, {long:.2f})"
        
        producer.produce(topic_name, key=message_key, value=message_value, callback=delivery_report)
        producer.poll(0.1)  # To trigger the delivery report callback for feedback

    producer.flush()  # Ensure all messages are sent

## Using producer.flush() before shutting down your producer ensures that all messages are delivered and that any callbacks associated with these messages are executed.