Merge pull request #2 from starichkov/protobuf-support

Added Protobuf support: extended producer and consumer to handle mult…

Author: starichkov

.github/workflows/python.yml

@@ -38,7 +38,7 @@ jobs:
       - name: Run tests with coverage
         run: |
           source .venv/bin/activate
-          pytest --cov=producer --cov=consumer --cov-report=xml
+          pytest --cov=producer --cov=consumer --cov=serialization --cov-report=xml
 
       # Only upload coverage from the main Python version (3.13)
       - name: Upload coverage to Codecov (only on Python 3.13)
@@ -65,7 +65,7 @@ jobs:
       - name: Build images
         run: docker compose build
 
-      - name: Start stack
+      - name: Start stack (JSON default)
         run: docker compose up -d
 
       - name: Wait for Kafka to be healthy
@@ -80,7 +80,7 @@ jobs:
           done
           echo "Kafka did not become healthy in time"; docker compose ps; docker compose logs kafka; exit 1
 
-      - name: Verify producer and consumer activity
+      - name: Verify producer and consumer activity (JSON)
         run: |
           # wait up to 120s for producer to send and consumer to log messages
           for i in $(seq 1 120); do
@@ -97,6 +97,58 @@ jobs:
           docker compose logs
           exit 1
 
+      - name: Tear down (after JSON run)
+        run: docker compose down -v --remove-orphans
+
+      - name: Start stack (Protobuf)
+        run: MESSAGE_FORMAT=protobuf docker compose up -d
+
+      - name: Wait for Kafka to be healthy (protobuf run)
+        run: |
+          for i in $(seq 1 60); do
+            status=$(docker inspect -f '{{json .State.Health.Status}}' kafka 2>/dev/null || echo "null")
+            echo "Kafka health: $status"
+            if [ "$status" = "\"healthy\"" ]; then
+              exit 0
+            fi
+            sleep 2
+          done
+          echo "Kafka did not become healthy in time"; docker compose ps; docker compose logs kafka; exit 1
+
+      - name: Verify producer and consumer activity (Protobuf)
+        run: |
+          # wait up to 120s for producer to send and consumer to log messages
+          for i in $(seq 1 120); do
+            prod=$(docker logs kafka-producer 2>&1 | grep -c 'Sent: key=' || true)
+            # Protobuf messages are logged as JSON after parsing
+            cons=$(docker logs kafka-consumer 2>&1 | grep -E -c 'JSON \(' || true)
+            echo "Attempt $i: producer=$prod consumer=$cons"
+            if [ "$prod" -ge 3 ] && [ "$cons" -ge 3 ]; then
+              echo "Producer and consumer logs look good (protobuf)"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "Producer/consumer (protobuf) did not show expected activity"
+          docker compose logs
+          exit 1
+
+      - name: Verify wire annotation appears (Protobuf)
+        run: |
+          # After activity is verified, ensure the consumer logs show at least one wire=protobuf annotation
+          for i in $(seq 1 60); do
+            ann=$(docker logs kafka-consumer 2>&1 | grep -c '\[wire=protobuf\]' || true)
+            echo "Attempt $i: wire_protobuf_annotations=$ann"
+            if [ "$ann" -ge 1 ]; then
+              echo "Found wire=protobuf annotation in consumer logs"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "wire=protobuf annotation not found in consumer logs"
+          docker logs kafka-consumer || true
+          exit 1
+
       - name: Show service logs (for debugging)
         if: failure()
         run: docker compose logs --no-color

README.md

@@ -5,11 +5,11 @@
 
 # Apache Kafka Python Demo
 
-## 🧩 Overview
+## Overview
 
 A minimal Apache Kafka demo using Python featuring:
-- A **JSON-producing** `producer.py`
-- A **format-tolerant** `consumer.py` supporting JSON & plain text
+- A multi-format `producer.py` (JSON by default; Protobuf via env)
+- A **format-tolerant** `consumer.py` supporting JSON, Protobuf & plain text
 - Partition-awareness, message keying, and consumer group support
 - Integration testing via `Testcontainers`
 - GitHub Actions & Codecov integration
@@ -18,7 +18,7 @@ It’s designed to simulate a polyglot messaging environment, where different sy
 
 ---
 
-## ⚙️ Requirements
+## Requirements
 
 - Python 3.10+
 - Docker & Docker Compose (for container-based setup) - more details could be found in the [separate section](documentation/docker.md).
@@ -32,7 +32,7 @@ pip install -r requirements.txt
 
 ---
 
-## 🚀 Running the Demo
+## Running the Demo
 
 ### 1. Start Apache Kafka
 
@@ -46,27 +46,49 @@ docker run -d --name kafka-391 \
 
 More information about helper scripts Kafka provides could be found in the [separate section](documentation/kafka.md).
 
-### 🟢 Run Producer
+### Run Producer
 
 ```bash
 python producer.py
 ```
 
-Produces random JSON events (`note_created`, `note_updated`, `note_deleted`) with message keys for partitioning.
+Produces random events (`note_created`, `note_updated`, `note_deleted`) with message keys for partitioning.
 
-### 🔵 Run Consumer
+Formats:
+- JSON (default)
+- Protobuf (enable via `MESSAGE_FORMAT=protobuf`)
+
+Examples:
+
+```bash
+# JSON (default)
+python producer.py
+
+# Protobuf
+MESSAGE_FORMAT=protobuf python producer.py
+```
+
+### Run Consumer
 
 ```bash
 python consumer.py                   # All events
 python consumer.py --event-type X   # Filtered by event_type
 python consumer.py --group-id my-group
 ```
 
-Displays event type, partition, and offset info.
+Displays event type, partition, and offset info. The consumer detects the payload format using the Kafka `content-type` header sent by the producer and falls back to JSON-or-plain-text when the header is missing.
+
+Log annotation:
+- Each consumed message line includes a suffix indicating the wire format detected: `[wire=protobuf]`, `[wire=json]`, `[wire=text]`, or `[wire=unknown]`.
+- Example:
+  - `✅ JSON (note_created) | key=note_created | partition=0 | offset=1 → {...} [wire=protobuf]`
+  - `📦 Plain | key=plain | partition=0 | offset=42 → hello [wire=text]`
+  
+Note: The `✅ JSON (...)` prefix reflects that the payload was parsed into a dict (even for Protobuf). The `[wire=...]` suffix shows the on-wire format.
 
 ---
 
-### 🔑 Message Keys and Partitions
+### Message Keys and Partitions
 
 The producer now uses the message's `event_type` as the Kafka **key**, which ensures that:
 
@@ -99,7 +121,7 @@ To see all partitions used, try increasing the number of unique keys or remove t
 
 ---
 
-### 👥 Consumer Groups and Partition Rebalancing
+### Consumer Groups and Partition Rebalancing
 
 Kafka uses consumer groups to distribute the workload of processing messages across multiple consumers.
 
@@ -123,7 +145,7 @@ Note: If you have more partitions than consumers, some consumers may receive mul
 
 ---
 
-## 🧪 Running Tests & Coverage
+## Running Tests & Coverage
 
 Details could be found in the [separate section](documentation/tests.md).
 
@@ -136,7 +158,7 @@ scripts/compose_smoke_test.sh --dry-run
 
 ---
 
-## 📂 Project Structure
+## Project Structure
 
 ```
 kafka-python-demo/
@@ -155,7 +177,7 @@ kafka-python-demo/
 
 ---
 
-## 📌 Notes
+## Notes
 
 - Topic name is configurable via environment variable `KAFKA_TOPIC` (default: `notes-topic`).
 - You can edit the scripts to change topic names or message structures.
@@ -165,6 +187,7 @@ kafka-python-demo/
 
 - `KAFKA_BOOTSTRAP_SERVERS` — Kafka broker(s), default: `localhost:9092`
 - `KAFKA_TOPIC` — topic to produce/consume, default: `notes-topic`
+- `MESSAGE_FORMAT` — producer payload format: `json` (default) or `protobuf`
 
 Examples:
 
@@ -176,11 +199,14 @@ python consumer.py
 
 # Using Docker Compose (host env is picked up by compose)
 KAFKA_TOPIC=my-topic docker compose up -d
+
+# Switch producer to Protobuf in Docker Compose
+MESSAGE_FORMAT=protobuf docker compose up -d
 ```
 
 ---
 
-## 🔗 Links
+## Links
 
 - [Status of Python versions](https://devguide.python.org/versions/)
 - [Apache Kafka (Official)](https://kafka.apache.org/)
@@ -192,15 +218,15 @@ KAFKA_TOPIC=my-topic docker compose up -d
 
 ---
 
-## 🧾 About TemplateTasks
+## About TemplateTasks
 
 TemplateTasks is a personal software development initiative by Vadim Starichkov, focused on sharing open-source libraries, services, and technical demos.
 
 It operates independently and outside the scope of any employment.
 
 All code is released under permissive open-source licenses. The legal structure may evolve as the project grows.
 
-## 📜 License & Attribution
+## License & Attribution
 
 This project is licensed under the **MIT License** - see the [LICENSE](https://github.com/starichkov/kafka-python-demo/blob/main/LICENSE.md) file for details.
 

consumer.py

@@ -9,7 +9,7 @@
 from kafka import KafkaConsumer
 from logger import get_logger
 import argparse
-from serialization import DESERIALIZERS
+from serialization import DESERIALIZERS, deserializer_for_mime
 
 
 def build_parser():
@@ -72,22 +72,67 @@ def consume_events(topic, consumer_args, event_type=None, group_id=None):
     try:
         # Continuously poll for new messages
         for message in consumer:
-            # Try to parse the message as JSON, fall back to plain text if not valid JSON
-            parsed = try_parse_json(message.value)
+            # Choose deserializer by content-type header if present; fallback to JSON-or-text
+            mime = None
+            try:
+                # headers: list[tuple[str, bytes]] (kafka-python)
+                if getattr(message, "headers", None):
+                    for k, v in message.headers:
+                        if (k == "content-type") or (isinstance(k, bytes) and k.decode("ascii", "ignore") == "content-type"):
+                            if isinstance(v, (bytes, bytearray)):
+                                mime = v.decode("ascii", errors="ignore")
+                            else:
+                                mime = str(v)
+                            break
+            except Exception:  # pragma: no cover
+                mime = None
+
+            try:
+                parser = deserializer_for_mime(mime)
+            except Exception:  # pragma: no cover
+                parser = DESERIALIZERS["json_or_text"]
+
+            # Determine wire format hint for logging
+            wire_hint = None
+            if mime:
+                mime_l = mime.lower()
+                if "json" in mime_l:
+                    wire_hint = "json"
+                elif "protobuf" in mime_l:
+                    wire_hint = "protobuf"
+                elif "text" in mime_l:
+                    wire_hint = "text"
+                else:
+                    wire_hint = "unknown"
+
+            # Parse the message value
+            try:
+                parsed = parser(message.value)
+            except Exception:
+                # As a last resort, fall back to text
+                parsed = DESERIALIZERS["plain_text"](message.value)
+
+            # If there was no content-type header, infer from parse result
+            if not mime:
+                if isinstance(parsed, dict):
+                    wire_hint = "json"
+                else:
+                    wire_hint = "text"
 
             # Decode key if available
             key = message.key.decode('utf-8') if message.key else None
             partition = message.partition
             offset = message.offset
 
             # Display the message with an appropriate prefix based on its type
+            suffix = f" [wire={wire_hint}]" if wire_hint else ""
             if isinstance(parsed, dict):
                 if event_type and parsed.get("event_type") != event_type:
                     continue  # Skip non-matching event
                 logger.info(
-                    f"✅ JSON ({parsed['event_type']}) | key={key} | partition={partition} | offset={offset} → {parsed}")
+                    f"✅ JSON ({parsed['event_type']}) | key={key} | partition={partition} | offset={offset} → {parsed}{suffix}")
             else:
-                logger.info(f"📦 Plain | key={key} | partition={partition} | offset={offset} → {parsed}")
+                logger.info(f"📦 Plain | key={key} | partition={partition} | offset={offset} → {parsed}{suffix}")
     except KeyboardInterrupt:
         # Handle graceful shutdown on Ctrl+C
         logger.info("\nShutting down gracefully...")

docker-compose.yml

@@ -47,6 +47,7 @@ services:
     environment:
       KAFKA_BOOTSTRAP_SERVERS: kafka:9092
       KAFKA_TOPIC: ${KAFKA_TOPIC:-notes-topic}
+      MESSAGE_FORMAT: ${MESSAGE_FORMAT:-json}
     restart: on-failure
 
   consumer: