From 81129056a0be89dbeaaaf25c359806f8aa0597dd Mon Sep 17 00:00:00 2001 From: John Patrick Teruel Date: Fri, 24 Apr 2026 21:18:15 +0800 Subject: [PATCH] Fix Kafka broker connection - use internal Docker port 29092 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixed two issues with Kafka connectivity: 1. Broker Address: Changed from 'localhost:9092' to 'kafka:29092' - localhost:9092 was trying to connect to host machine, not Docker container - kafka:29092 is the internal Docker network listener configured in docker-compose - This allows the API container to reach Kafka via Docker's internal DNS 2. Kafka Producer Config: Removed invalid 'api_version_auto_discovery_interval_ms' - Replaced with 'api_version=(2, 5, 0)' for explicit API version - Confluent Kafka 7.5.0 uses API v2.5 - Prevents unrecognized config warnings Docker-Compose Kafka Configuration: KAFKA_ADVERTISED_LISTENERS: - PLAINTEXT://kafka:29092 (for Docker internal communication) - PLAINTEXT_HOST://localhost:9092 (for host machine access) Result: Transactions now successfully published to Kafka! POST /transaction → 200 OK with transaction_id Message broker available and working Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- api/main.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/api/main.py b/api/main.py index bc56c8e..fdd9303 100644 --- a/api/main.py +++ b/api/main.py @@ -32,8 +32,8 @@ import logging SECRET_KEY = "payguard-demo-secret-key-change-in-production" ALGORITHM = "HS256" -# Kafka bootstrap server -KAFKA_BROKER = "localhost:9092" +# Kafka bootstrap server (use 'kafka:29092' for Docker internal, localhost:9092 for host access) +KAFKA_BROKER = "kafka:29092" KAFKA_TOPIC = "transactions" # Logging setup @@ -75,8 +75,8 @@ def get_kafka_producer(): value_serializer=lambda v: json.dumps(v).encode('utf-8'), acks='all', retries=3, - api_version_auto_discovery_interval_ms=5000, - request_timeout_ms=10000 + request_timeout_ms=10000, + api_version=(2, 5, 0) # Specify Kafka API version ) logger.info(f"[KAFKA PRODUCER] Initialized, broker={KAFKA_BROKER}, topic={KAFKA_TOPIC}") except Exception as e: