pat/payguard-poc
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix Kafka broker connection - use internal Docker port 29092

Browse Source 
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 <[email protected]>
This commit is contained in:
John Patrick TeruelandCopilot committed 2026-04-24 21:18:15 +08:00
1 parent bb7af74c77
commit 81129056a0
1 file changed
+4 -4
Download Patch File Download Diff File Expand all files Collapse all files
api/main.py
+4 -4
View File
@@ -32,8 +32,8 @@ import logging
SECRET_KEY = "payguard-demo-secret-key-change-in-production" SECRET_KEY = "payguard-demo-secret-key-change-in-production"
ALGORITHM = "HS256" ALGORITHM = "HS256"
# Kafka bootstrap server # Kafka bootstrap server (use 'kafka:29092' for Docker internal, localhost:9092 for host access)
KAFKA_BROKER = "localhost:9092" KAFKA_BROKER = "kafka:29092"
KAFKA_TOPIC = "transactions" KAFKA_TOPIC = "transactions"
# Logging setup # Logging setup
@@ -75,8 +75,8 @@ def get_kafka_producer():
value_serializer=lambda v: json.dumps(v).encode('utf-8'), value_serializer=lambda v: json.dumps(v).encode('utf-8'),
acks='all', acks='all',
retries=3, 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}") logger.info(f"[KAFKA PRODUCER] Initialized, broker={KAFKA_BROKER}, topic={KAFKA_TOPIC}")
except Exception as e: except Exception as e: