muffin-kafka 2.1.2

Kafka Integration for Muffin framework

Muffin-Kafka

Muffin-Kafka is an Apache Kafka integration plugin for the Muffin web framework, built on top of aiokafka.

---

🚀 Features

• Async Kafka integration using aiokafka
• Single or batch message consumption — stream messages one-by-one or read in batches
• Per-topic task model — each topic is consumed in an isolated asyncio task
• Simple handler registration using @plugin.handle_topics(...)
• Manual or auto-commit support, custom group IDs
• Producer support via send
• Built-in monitoring with offsets, lag, and poll delay
• Healthcheck manage command for liveness probes and observability
• Optional error handler via @plugin.handle_error(...)

---

✅ Requirements

• Python ≥ 3.10
• Muffin ≥ 0.71
• Kafka cluster or broker (local or cloud)

---

📦 Installation

pip install muffin-kafka

⚙️ Usage

    from muffin import Application
    from muffin_kafka import Kafka

    app = Application("example")

    # Initialize the plugin with config options
    kafka = Kafka(app, bootstrap_servers="localhost:9092", produce=True, listen=True)

🧩 Registering Handlers

Use @kafka.handle_topics(...) to register a handler for specific Kafka topics:

    @kafka.handle_topics("events.user", "events.auth")
    async def handle_event(message):
        data = message.value.decode()
        print("Received:", data)

You can also register a global error handler:

    @kafka.handle_error
    async def on_error(exc):
        print("Kafka error:", exc)

📤 Sending Messages

You can send messages to Kafka topics using the send method:

    # Send a message
    await kafka.send("events.user", {"action": "signup"}, key="user123")

🔄 Healthcheck

Run the kafka-healthcheck management command to check consumer lag:

    python app.py kafka-healthcheck

Or programmatically via Muffin's manage command:

    # Check specific topics
    await app.manage.commands["kafka-healthcheck"]("events.user", max_lag=1000)

📊 Monitoring

If monitor=True is passed, the plugin will log:

• Committed offsets
• Latest offsets
• Poll timestamps
• Per-partition lag and delay

This data can be extended for Prometheus/Grafana metrics or alerting.

⚙️ Configuration Options

You can pass configuration options either as keyword arguments to the plugin:

kafka = Kafka(app, bootstrap_servers="localhost:9092", produce=True)

Or set them via Muffin's config system (e.g. .env, YAML):

"KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
"KAFKA_PRODUCE": True,

Supported Options

Option	Type	Default	Description
bootstrap_servers	str	"localhost:9092"	Kafka broker connection string
group_id	str	None	Kafka consumer group ID
client_id	str	"muffin"	Kafka client ID
produce	bool	False	Enable Kafka producer
listen	bool	True	Enable consumers (message listening)
monitor	bool	False	Enable internal consumer monitor
batch_size	int	None	Read messages in batches (uses getmany())
monitor_interval	int	60	Monitor frequency in seconds
auto_offset_reset	str	"earliest"	Where to start if no committed offset
enable_auto_commit	bool	False	Auto-commit offsets after each message/batch. When False, the plugin commits manually after all handlers succeed.
max_poll_records	int	None	Max records to poll in one batch
request_timeout_ms	int	30000	Request timeout
retry_backoff_ms	int	1000	Retry interval on failure
security_protocol	str	"PLAINTEXT"	Kafka protocol (SSL, SASL_PLAINTEXT, …)
sasl_mechanism	str	"PLAIN"	SASL auth mechanism
sasl_plain_username	str	None	SASL auth user
sasl_plain_password	str	None	SASL auth password
ssl_cafile	str	None	Path to trusted CA certs

---

🐞 Bug Tracker

Found a bug or have a feature request? Please open an issue at: https://github.com/klen/muffin-kafka/issues

---

🤝 Contributing

Pull requests are welcome! Development happens here: https://github.com/klen/muffin-kafka

---

🪪 License

MIT – See LICENSE for full details.