NoBrokersAvailable error when running "How to connect" sample python code
Last updated: August 10, 2026
Issue
When using the Python sample code found in
Redpanda Cloud Console :Overview > Kafka API (How to Connect) > Python
Redpanda Cloud Console :Overview > Kafka API (How to Connect) > Python
the producer/consumer code was failing with error
kafka.errors.NoBrokersAvailable: NoBrokersAvailable
Detail
kafka.errors.NoBrokersAvailable: NoBrokersAvailable is a fairly generic error which could be multiple causes
Such as :
- Incorrect SASL info - username/password/sasl-mechanism
-missing ACLS for user
- TLS/SSL Handshake issue (missing CA certs for example )
Resolution
If you encounter kafka.errors.NoBrokersAvailable: NoBrokersAvailable and it is not a clear what the cause is, then you can use a revised version of the producer code with enhanced logging and error checking to help troubleshoot .
consumer
import sys
import logging
from kafka import KafkaAdminClient
from kafka.admin import NewTopic
from kafka.errors import (
TopicAlreadyExistsError,
KafkaError,
NoBrokersAvailable,
KafkaConnectionError,
AuthenticationFailedError,
IllegalSaslStateError,
KafkaTimeoutError,
TopicAuthorizationFailedError,
ClusterAuthorizationFailedError,
)
# Custom handler to intercept kafka client logs
class AuthErrorDetector(logging.Handler):
def __init__(self):
super().__init__()
self.auth_error_detected = False
self.auth_success_detected = False
self.ssl_error_detected = False
self.auth_error_messages = []
self.auth_success_messages = []
self.ssl_error_messages = []
def emit(self, record):
msg = record.getMessage()
msg_lower = msg.lower()
# Detect SSL/TLS handshake failures
if '' in msg and ('ssl connection closed' in msg_lower or
'closing connection' in msg_lower):
self.ssl_error_detected = True
self.ssl_error_messages.append(msg)
# Also catch SSL warnings
elif 'ssl connection closed' in msg_lower and 'handshake' in msg_lower:
self.ssl_error_detected = True
self.ssl_error_messages.append(msg)
# Detect authentication SUCCESS
elif 'authenticated as' in msg_lower and 'via scram' in msg_lower:
self.auth_success_detected = True
self.auth_success_messages.append(msg)
# Detect authentication failures
elif '' in msg and any(keyword in msg_lower for keyword in
['connection reset', 'error receiving', 'closing connection']):
self.auth_error_detected = True
self.auth_error_messages.append(msg)
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Set up kafka logger with our custom handler
kafka_logger = logging.getLogger('kafka')
kafka_logger.setLevel(logging.INFO)
auth_detector = AuthErrorDetector()
kafka_logger.addHandler(auth_detector)
# Configuration
BOOTSTRAP_SERVERS = ""
SASL_MECHANISM = "SCRAM-SHA-256" # or "SCRAM-SHA-512"
USERNAME = ""
PASSWORD = ""
TOPIC_NAME = "demo-topic"
NUM_PARTITIONS = 1
REPLICATION_FACTOR = -1 # -1 means use cluster default
def validate_config():
"""Validate configuration before attempting connection"""
issues = []
if "" in USERNAME or not USERNAME:
issues.append("❌ Username is not set (still contains placeholder)")
if "" in PASSWORD or not PASSWORD:
issues.append("❌ Password is not set (still contains placeholder)")
if SASL_MECHANISM not in ["SCRAM-SHA-256", "SCRAM-SHA-512"]:
issues.append(f"❌ Invalid SASL mechanism: {SASL_MECHANISM}")
if issues:
logger.error("Configuration validation failed:")
for issue in issues:
logger.error(f" {issue}")
return False
logger.info("✓ Configuration validation passed")
return True
def create_admin_client():
"""Create Kafka admin client with detailed error handling"""
logger.info("Connecting to Redpanda...")
logger.info(f" Bootstrap server: {BOOTSTRAP_SERVERS}")
logger.info(f" SASL mechanism: {SASL_MECHANISM}")
logger.info(f" Username: {USERNAME}")
# Reset auth detector
auth_detector.auth_error_detected = False
auth_detector.auth_success_detected = False
auth_detector.ssl_error_detected = False
auth_detector.auth_error_messages = []
auth_detector.auth_success_messages = []
auth_detector.ssl_error_messages = []
try:
admin = KafkaAdminClient(
bootstrap_servers=BOOTSTRAP_SERVERS,
security_protocol="SASL_SSL",
sasl_mechanism=SASL_MECHANISM,
sasl_plain_username=USERNAME,
sasl_plain_password=PASSWORD,
# Shorter timeouts to fail faster
request_timeout_ms=10000,
api_version_auto_timeout_ms=5000,
)
# Test connection by listing topics
logger.info("Testing connection and permissions...")
try:
topics = admin.list_topics()
logger.info(f"✓ Successfully connected! Found {len(topics)} topics in cluster")
except Exception as test_error:
raise
logger.info("✓ Admin client ready")
return admin
except Exception as e:
# First priority: Check for SSL/TLS errors
if auth_detector.ssl_error_detected:
logger.error("=" * 70)
logger.error("❌ SSL/TLS HANDSHAKE FAILED")
logger.error("=" * 70)
logger.error("")
logger.error(" The SSL/TLS connection failed during the handshake phase.")
logger.error("")
logger.error(" ⚠️ MOST COMMON CAUSE:")
logger.error(" 🔴 MISSING OR INVALID SSL CERTIFICATES")
logger.error("")
logger.error(" ⚠️ HOW TO FIX:")
logger.error("")
logger.error(" Option 1 - Install system CA certificates (recommended):")
logger.error(" • macOS: brew install ca-certificates && brew link ca-certificates")
logger.error(" • Ubuntu/Debian: apt-get install ca-certificates")
logger.error("")
logger.error(" Option 2 - Specify CA certificate explicitly:")
logger.error(" admin = KafkaAdminClient(")
logger.error(" ...,")
logger.error(" ssl_cafile='/path/to/ca-cert.pem',")
logger.error(" )")
logger.error("")
return None
# Second: Check if authentication succeeded
if auth_detector.auth_success_detected:
logger.error("=" * 70)
logger.error("✓ Authentication Successful → BUT Access Denied")
logger.error("=" * 70)
logger.error("")
logger.error(f" ✓ Successfully authenticated as: '{USERNAME}'")
logger.error(" ✓ SSL/TLS handshake completed")
logger.error(" ✓ Connected to broker")
logger.error("")
# Determine what went wrong AFTER successful auth
if isinstance(e, (KafkaTimeoutError, ClusterAuthorizationFailedError)):
logger.error(" ✗ Failed to list topics / access cluster metadata")
logger.error("")
logger.error("=" * 70)
logger.error("❌ INSUFFICIENT PERMISSIONS (ACL ISSUE)")
logger.error("=" * 70)
logger.error("")
logger.error(" What this means:")
logger.error(f" User '{USERNAME}' authenticated successfully BUT lacks")
logger.error(" permissions to perform admin operations.")
logger.error("")
logger.error(" 🔴 REQUIRED PERMISSIONS FOR ADMIN CLIENT:")
logger.error(f" User '{USERNAME}' needs:")
logger.error("")
logger.error(" 1. DESCRIBE on CLUSTER (to list topics)")
logger.error(" 2. CREATE on CLUSTER or TOPIC (to create topics)")
logger.error(" 3. DESCRIBE on TOPIC (to get topic details)")
logger.error(" 4. ALTER on TOPIC (to modify topic configs)")
logger.error(" 5. DELETE on TOPIC (to delete topics)")
logger.error("")
logger.error(" ⚠️ HOW TO FIX:")
logger.error("")
logger.error(" Step 1: Log into Redpanda Console")
logger.error("")
logger.error(" Step 2: Navigate to Security → ACLs")
logger.error("")
logger.error(f" Step 3: Grant cluster-level permissions to '{USERNAME}':")
logger.error("")
logger.error(" Resource Type: CLUSTER")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: DESCRIBE, CREATE")
logger.error("")
logger.error(" Step 4: Grant topic-level permissions (for topic management):")
logger.error("")
logger.error(" Resource Type: TOPIC")
logger.error(" Resource Name: * (all topics) or specific topic name")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: DESCRIBE, CREATE, ALTER, DELETE")
logger.error("")
logger.error(" Alternative (CLI):")
logger.error(f" # Grant cluster permissions")
logger.error(f" rpk security acl create --allow-principal User:{USERNAME} \\")
logger.error(" --operation describe,create --cluster")
logger.error("")
logger.error(f" # Grant topic permissions")
logger.error(f" rpk security acl create --allow-principal User:{USERNAME} \\")
logger.error(" --operation describe,create,alter,delete --topic '*'")
logger.error("")
else:
logger.error(f"❌ UNEXPECTED ERROR AFTER AUTHENTICATION: {type(e).__name__}")
logger.error(f" Error: {e}")
return None
# Third: Authentication failed
if auth_detector.auth_error_detected:
logger.error("=" * 70)
logger.error("❌ AUTHENTICATION FAILED - INVALID CREDENTIALS")
logger.error("=" * 70)
logger.error("")
logger.error(" The broker rejected your username or password.")
logger.error("")
logger.error(" ⚠️ ACTION REQUIRED:")
logger.error(f" → Verify username: '{USERNAME}'")
logger.error(" → Verify password (check for typos)")
logger.error(" → Confirm credentials in Redpanda Console")
logger.error("")
logger.error(" Other possible causes:")
logger.error(f" • Wrong SASL mechanism (currently: {SASL_MECHANISM})")
if SASL_MECHANISM == "SCRAM-SHA-256":
logger.error(" Try: SASL_MECHANISM = 'SCRAM-SHA-512'")
else:
logger.error(" Try: SASL_MECHANISM = 'SCRAM-SHA-256'")
logger.error("")
return None
# Handle other specific exceptions
if isinstance(e, AuthenticationFailedError):
logger.error("=" * 70)
logger.error("❌ AUTHENTICATION FAILED")
logger.error("=" * 70)
logger.error(f" Your credentials were rejected: {e}")
return None
elif isinstance(e, IllegalSaslStateError):
logger.error("=" * 70)
logger.error("❌ SASL CONFIGURATION ERROR")
logger.error("=" * 70)
logger.error(f" Wrong SASL mechanism: {SASL_MECHANISM}")
logger.error(f" Error: {e}")
return None
elif isinstance(e, KafkaTimeoutError):
logger.error("=" * 70)
logger.error("❌ CONNECTION TIMEOUT")
logger.error("=" * 70)
logger.error(" Could not connect within timeout period")
logger.error("")
logger.error(" Note: If authentication succeeded (check logs above),")
logger.error(" this is most likely an ACL permissions issue.")
logger.error(f" Error: {e}")
return None
elif isinstance(e, NoBrokersAvailable):
logger.error("=" * 70)
logger.error("❌ NO BROKERS AVAILABLE")
logger.error("=" * 70)
logger.error(" Could not connect to any broker")
logger.error("")
logger.error(" Possible causes:")
logger.error(" 1. SSL/TLS certificate issue")
logger.error(" 2. Wrong bootstrap server address")
logger.error(" 3. Authentication failure")
logger.error(" 4. Network connectivity issues")
logger.error(f" Error: {e}")
return None
else:
logger.error("=" * 70)
logger.error(f"❌ UNEXPECTED ERROR: {type(e).__name__}")
logger.error("=" * 70)
logger.error(f" Error: {e}")
return None
def create_topic(admin, topic_name, num_partitions, replication_factor):
"""Create a topic with detailed error handling"""
logger.info(f"Creating topic '{topic_name}'...")
logger.info(f" Partitions: {num_partitions}")
logger.info(f" Replication factor: {replication_factor} (-1 = cluster default)")
try:
topic = NewTopic(
name=topic_name,
num_partitions=num_partitions,
replication_factor=replication_factor,
replica_assignments=[]
)
result = admin.create_topics(new_topics=[topic], validate_only=False)
# Check if creation was successful
logger.info(f"✓ Topic '{topic_name}' created successfully!")
return True
except TopicAlreadyExistsError:
logger.warning(f"⚠️ Topic '{topic_name}' already exists")
logger.info(" This is not an error - the topic is ready to use")
return True
except TopicAuthorizationFailedError as e:
logger.error("=" * 70)
logger.error("❌ TOPIC AUTHORIZATION FAILED")
logger.error("=" * 70)
logger.error("")
logger.error(f" User '{USERNAME}' lacks permission to CREATE topics")
logger.error("")
logger.error(" 🔴 REQUIRED PERMISSIONS:")
logger.error("")
logger.error(" Option 1 - Grant CREATE on specific topic:")
logger.error(" Resource Type: TOPIC")
logger.error(f" Resource Name: {topic_name}")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: CREATE, DESCRIBE")
logger.error("")
logger.error(" Option 2 - Grant CREATE on all topics:")
logger.error(" Resource Type: TOPIC")
logger.error(" Resource Name: * (wildcard)")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: CREATE, DESCRIBE")
logger.error("")
logger.error(" Option 3 - Grant CREATE at cluster level:")
logger.error(" Resource Type: CLUSTER")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: CREATE")
logger.error("")
logger.error(" CLI command:")
logger.error(f" rpk security acl create --allow-principal User:{USERNAME} \\")
logger.error(f" --operation create,describe --topic {topic_name}")
logger.error("")
logger.error(f" Error details: {e}")
return False
except ClusterAuthorizationFailedError as e:
logger.error("=" * 70)
logger.error("❌ CLUSTER AUTHORIZATION FAILED")
logger.error("=" * 70)
logger.error("")
logger.error(f" User '{USERNAME}' lacks cluster-level CREATE permission")
logger.error("")
logger.error(" ⚠️ HOW TO FIX:")
logger.error(" Grant cluster-level CREATE permission:")
logger.error("")
logger.error(" Resource Type: CLUSTER")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: CREATE")
logger.error("")
logger.error(" CLI command:")
logger.error(f" rpk security acl create --allow-principal User:{USERNAME} \\")
logger.error(" --operation create --cluster")
logger.error("")
logger.error(f" Error details: {e}")
return False
except KafkaError as e:
error_str = str(e)
logger.error("=" * 70)
logger.error(f"❌ KAFKA ERROR: {type(e).__name__}")
logger.error("=" * 70)
logger.error(f" Error: {e}")
if "authorization" in error_str.lower() or "permission" in error_str.lower():
logger.error("")
logger.error(" This appears to be a permissions issue.")
logger.error(f" User '{USERNAME}' may need additional ACL permissions.")
return False
except Exception as e:
logger.error("=" * 70)
logger.error(f"❌ UNEXPECTED ERROR: {type(e).__name__}")
logger.error("=" * 70)
logger.error(f" Error: {e}")
return False
def main():
logger.info("=" * 70)
logger.info("Redpanda Admin Client - Topic Creation")
logger.info("=" * 70)
# Validate configuration
if not validate_config():
logger.error("\n⚠️ Fix configuration issues above")
sys.exit(1)
# Create admin client
admin = create_admin_client()
if not admin:
logger.error("")
logger.error("=" * 70)
logger.error("⚠️ CONNECTION FAILED - See error details above")
logger.error("=" * 70)
sys.exit(1)
try:
# Create topic
success = create_topic(admin, TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR)
if success:
logger.info("")
logger.info("=" * 70)
logger.info("✓ OPERATION COMPLETED SUCCESSFULLY")
logger.info("=" * 70)
logger.info(f" Topic '{TOPIC_NAME}' is ready to use")
# Optionally, describe the topic
try:
topics_metadata = admin.describe_topics([TOPIC_NAME])
if topics_metadata:
topic_info = topics_metadata[0]
logger.info(f" Partitions: {len(topic_info['partitions'])}")
logger.info(f" Replication factor: {len(topic_info['partitions'][0]['replicas']) if topic_info['partitions'] else 'N/A'}")
except Exception:
pass # Optional, don't fail if we can't describe
sys.exit(0)
else:
logger.error("")
logger.error("=" * 70)
logger.error("⚠️ OPERATION FAILED - See error details above")
logger.error("=" * 70)
sys.exit(1)
except KeyboardInterrupt:
logger.warning("\n⚠️ Interrupted by user")
sys.exit(1)
except Exception as e:
logger.error(f"❌ Unexpected error: {e}")
sys.exit(1)
finally:
logger.info("\nClosing admin client...")
admin.close()
logger.info("✓ Admin client closed")
if __name__ == "__main__":
main()
producer
import socket
import sys
import logging
from kafka import KafkaProducer
from kafka.errors import (
KafkaError,
NoBrokersAvailable,
KafkaConnectionError,
AuthenticationFailedError,
IllegalSaslStateError,
KafkaTimeoutError,
)
# Custom handler to intercept kafka client logs
class AuthErrorDetector(logging.Handler):
def __init__(self):
super().__init__()
self.auth_error_detected = False
self.auth_success_detected = False
self.ssl_error_detected = False
self.auth_error_messages = []
self.auth_success_messages = []
self.ssl_error_messages = []
def emit(self, record):
msg = record.getMessage()
msg_lower = msg.lower()
# Detect SSL/TLS handshake failures
if '' in msg and ('ssl connection closed' in msg_lower or
'closing connection' in msg_lower):
self.ssl_error_detected = True
self.ssl_error_messages.append(msg)
# Also catch SSL warnings
elif 'ssl connection closed' in msg_lower and 'handshake' in msg_lower:
self.ssl_error_detected = True
self.ssl_error_messages.append(msg)
# Detect authentication SUCCESS
elif 'authenticated as' in msg_lower and 'via scram' in msg_lower:
self.auth_success_detected = True
self.auth_success_messages.append(msg)
# Detect authentication failures
elif '' in msg and any(keyword in msg_lower for keyword in
['connection reset', 'error receiving', 'closing connection']):
self.auth_error_detected = True
self.auth_error_messages.append(msg)
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Set up kafka logger with our custom handler
kafka_logger = logging.getLogger('kafka')
kafka_logger.setLevel(logging.INFO)
auth_detector = AuthErrorDetector()
kafka_logger.addHandler(auth_detector)
# Configuration
BOOTSTRAP_SERVERS = ""
SASL_MECHANISM = "SCRAM-SHA-256" # or "SCRAM-SHA-512"
USERNAME = ""
PASSWORD = ""
TOPIC = "demo-topic"
def validate_config():
"""Validate configuration before attempting connection"""
issues = []
if "" in USERNAME or not USERNAME:
issues.append("❌ Username is not set (still contains placeholder)")
if "" in PASSWORD or not PASSWORD:
issues.append("❌ Password is not set (still contains placeholder)")
if SASL_MECHANISM not in ["SCRAM-SHA-256", "SCRAM-SHA-512"]:
issues.append(f"❌ Invalid SASL mechanism: {SASL_MECHANISM}")
if issues:
logger.error("Configuration validation failed:")
for issue in issues:
logger.error(f" {issue}")
return False
logger.info("✓ Configuration validation passed")
return True
def create_producer():
"""Create Kafka producer with detailed error handling"""
logger.info("Connecting to Redpanda...")
logger.info(f" Bootstrap server: {BOOTSTRAP_SERVERS}")
logger.info(f" SASL mechanism: {SASL_MECHANISM}")
logger.info(f" Username: {USERNAME}")
# Reset auth detector
auth_detector.auth_error_detected = False
auth_detector.auth_success_detected = False
auth_detector.ssl_error_detected = False
auth_detector.auth_error_messages = []
auth_detector.auth_success_messages = []
auth_detector.ssl_error_messages = []
try:
producer = KafkaProducer(
bootstrap_servers=BOOTSTRAP_SERVERS,
security_protocol="SASL_SSL",
sasl_mechanism=SASL_MECHANISM,
sasl_plain_username=USERNAME,
sasl_plain_password=PASSWORD,
# Shorter timeouts to fail faster
request_timeout_ms=10000,
api_version_auto_timeout_ms=5000,
connections_max_idle_ms=10000,
max_block_ms=10000,
)
# Force connection attempt
logger.info("Fetching topic metadata...")
try:
topics = producer.partitions_for(TOPIC)
if topics is None:
logger.warning(f"⚠️ Topic '{TOPIC}' doesn't exist")
logger.warning(" It will be auto-created on first message (if allowed)")
else:
logger.info(f"✓ Topic '{TOPIC}' found with {len(topics)} partitions")
except Exception as metadata_error:
raise
logger.info("✓ Producer ready")
return producer
except Exception as e:
# First priority: Check for SSL/TLS errors
if auth_detector.ssl_error_detected:
logger.error("=" * 70)
logger.error("❌ SSL/TLS HANDSHAKE FAILED")
logger.error("=" * 70)
logger.error("")
logger.error(" The SSL/TLS connection failed during the handshake phase.")
logger.error(" The server closed the connection before encryption could be established.")
logger.error("")
logger.error(" What happened:")
logger.error(" 1. Connected to broker successfully ✓")
logger.error(" 2. Started SSL/TLS handshake...")
logger.error(" 3. Server rejected SSL connection ✗")
logger.error(" 4. Connection closed ✗")
logger.error("")
logger.error(" ⚠️ MOST COMMON CAUSES:")
logger.error("")
logger.error(" 1. 🔴 MISSING OR INVALID SSL CERTIFICATES")
logger.error(" • System CA certificates are not configured properly")
logger.error(" • Wrong certificate path")
logger.error(" • Certificates are expired or corrupted")
logger.error("")
logger.error(" 2. Server requires client certificate (mutual TLS)")
logger.error(" • You may need to provide ssl_certfile and ssl_keyfile")
logger.error("")
logger.error(" 3. SSL protocol version mismatch")
logger.error(" • Server may require specific TLS version")
logger.error("")
logger.error(" ⚠️ HOW TO FIX:")
logger.error("")
logger.error(" Option 1 - Use system CA certificates (recommended):")
logger.error(" • macOS: brew install ca-certificates && brew link ca-certificates")
logger.error(" • Ubuntu/Debian: apt-get install ca-certificates")
logger.error(" • Ensure /etc/ssl/certs/ or similar path exists")
logger.error("")
logger.error(" Option 2 - Specify CA certificate explicitly:")
logger.error(" producer = KafkaProducer(")
logger.error(" ...,")
logger.error(" ssl_cafile='/path/to/ca-cert.pem',")
logger.error(" )")
logger.error("")
logger.error(" Option 3 - For testing ONLY (insecure, not recommended):")
logger.error(" producer = KafkaProducer(")
logger.error(" ...,")
logger.error(" ssl_check_hostname=False,")
logger.error(" ssl_context=ssl.create_default_context(),")
logger.error(" )")
logger.error("")
logger.error(" Current SSL CA paths being checked:")
for msg in auth_detector.ssl_error_messages[:2]:
if 'DefaultVerifyPaths' in msg or 'cafile' in msg.lower():
logger.error(f" {msg}")
logger.error("")
return None
# Second: Check if authentication actually succeeded
if auth_detector.auth_success_detected:
logger.error("=" * 70)
logger.error("✓ Authentication Successful → BUT Access Denied")
logger.error("=" * 70)
logger.error("")
logger.error(f" ✓ Successfully authenticated as: '{USERNAME}'")
logger.error(" ✓ SSL/TLS handshake completed")
logger.error(" ✓ Connected to broker")
logger.error("")
# Determine what went wrong AFTER successful auth
if isinstance(e, KafkaTimeoutError):
logger.error(" ✗ Timed out fetching metadata")
logger.error("")
logger.error("=" * 70)
logger.error("❌ INSUFFICIENT PERMISSIONS (ACL ISSUE)")
logger.error("=" * 70)
logger.error("")
logger.error(" What this means:")
logger.error(f" User '{USERNAME}' authenticated successfully BUT lacks")
logger.error(" the necessary permissions to perform the requested operation.")
logger.error("")
logger.error(" What happened:")
logger.error(" 1. Connected to broker ✓")
logger.error(" 2. SSL/TLS handshake ✓")
logger.error(" 3. Authentication succeeded ✓")
logger.error(" 4. Requested metadata for topic...")
logger.error(" 5. ✗ NO RESPONSE (blocked by ACLs)")
logger.error(" 6. ✗ Operation timed out")
logger.error("")
logger.error(" Why metadata request failed:")
logger.error(f" • User '{USERNAME}' lacks DESCRIBE permission on topic '{TOPIC}'")
logger.error(" • OR user lacks DESCRIBE permission on cluster")
logger.error(" • ACLs are configured but this user is not authorized")
logger.error("")
logger.error(" 🔴 REQUIRED PERMISSIONS:")
logger.error(f" To produce messages to topic '{TOPIC}', user '{USERNAME}' needs:")
logger.error("")
logger.error(f" 1. DESCRIBE on topic '{TOPIC}' (to fetch metadata)")
logger.error(f" 2. WRITE on topic '{TOPIC}' (to produce messages)")
logger.error("")
logger.error(" Optional but recommended:")
logger.error(" 3. CREATE on topic (if topic doesn't exist)")
logger.error("")
logger.error(" ⚠️ HOW TO FIX:")
logger.error("")
logger.error(" Step 1: Log into Redpanda Console")
logger.error(" (https://your-cluster.redpanda.com)")
logger.error("")
logger.error(" Step 2: Navigate to Security → ACLs")
logger.error("")
logger.error(f" Step 3: Add ACL for user '{USERNAME}':")
logger.error("")
logger.error(" Resource Type: TOPIC")
logger.error(f" Resource Name: {TOPIC} (or * for all topics)")
logger.error(" Pattern Type: LITERAL")
logger.error(f" Principal: User:{USERNAME}")
logger.error(" Operations: DESCRIBE, WRITE")
logger.error("")
logger.error(" Alternative (CLI):")
logger.error(f" rpk security acl create --allow-principal User:{USERNAME} \\")
logger.error(f" --operation describe,write --topic {TOPIC}")
logger.error("")
logger.error(" After granting permissions, try running this script again.")
logger.error("")
else:
logger.error(f"❌ UNEXPECTED ERROR AFTER AUTHENTICATION: {type(e).__name__}")
logger.error(f" Error: {e}")
logger.error("")
logger.error(" Authentication succeeded, but something else failed.")
logger.error(" This might be a permissions issue or network problem.")
return None
# Third: Authentication failed - check if we detected the failure
if auth_detector.auth_error_detected:
logger.error("=" * 70)
logger.error("❌ AUTHENTICATION FAILED - INVALID CREDENTIALS")
logger.error("=" * 70)
logger.error("")
logger.error(" The broker rejected your username or password.")
logger.error("")
logger.error(" What happened:")
logger.error(" 1. Connected to broker successfully ✓")
logger.error(" 2. SSL/TLS handshake completed ✓")
logger.error(" 3. Started SASL authentication...")
logger.error(" 4. Broker rejected credentials ✗")
logger.error(" 5. Broker closed the connection ✗")
logger.error("")
logger.error(" ⚠️ ACTION REQUIRED:")
logger.error(f" → Verify username: '{USERNAME}'")
logger.error(" → Verify password (check for typos)")
logger.error(" → Confirm credentials in Redpanda Console")
logger.error("")
logger.error(" Other possible causes:")
logger.error(f" • Wrong SASL mechanism (currently: {SASL_MECHANISM})")
if SASL_MECHANISM == "SCRAM-SHA-256":
logger.error(" Try: SASL_MECHANISM = 'SCRAM-SHA-512'")
else:
logger.error(" Try: SASL_MECHANISM = 'SCRAM-SHA-256'")
logger.error(" • User account doesn't exist")
logger.error(" • User account is disabled")
logger.error("")
return None
# Handle other specific exceptions (when error type is unclear)
if isinstance(e, AuthenticationFailedError):
logger.error("=" * 70)
logger.error("❌ AUTHENTICATION FAILED")
logger.error("=" * 70)
logger.error(f" Your credentials were rejected: {e}")
return None
elif isinstance(e, IllegalSaslStateError):
logger.error("=" * 70)
logger.error("❌ SASL CONFIGURATION ERROR")
logger.error("=" * 70)
logger.error(f" Wrong SASL mechanism: {SASL_MECHANISM}")
logger.error(f" Error: {e}")
return None
elif isinstance(e, KafkaTimeoutError):
logger.error("=" * 70)
logger.error("❌ CONNECTION TIMEOUT")
logger.error("=" * 70)
logger.error(" Could not connect to broker within timeout period")
logger.error("")
logger.error(" Possible causes:")
logger.error(" 1. ACL/permissions issue (if auth succeeded)")
logger.error(" 2. Invalid credentials (auth failed silently)")
logger.error(" 3. Network/firewall blocking connection")
logger.error(" 4. Broker is unreachable")
logger.error("")
logger.error(" Note: Check if authentication succeeded in logs above.")
logger.error(" If yes, this is most likely an ACL permissions issue.")
logger.error(f" Error: {e}")
return None
elif isinstance(e, NoBrokersAvailable):
logger.error("=" * 70)
logger.error("❌ NO BROKERS AVAILABLE")
logger.error("=" * 70)
logger.error(" Could not connect to any broker")
logger.error("")
logger.error(" Possible causes:")
logger.error(" 1. SSL/TLS certificate issue (see errors above)")
logger.error(" 2. Wrong bootstrap server address")
logger.error(" 3. Authentication failure")
logger.error(" 4. Network connectivity issues")
logger.error(" 5. Broker is down")
logger.error("")
logger.error(" Troubleshooting:")
logger.error(f" • Test DNS: ping {BOOTSTRAP_SERVERS.split(':')[0]}")
logger.error(f" • Test port: nc -zv {BOOTSTRAP_SERVERS.split(':')[0]} 9092")
logger.error(f" Error: {e}")
return None
elif isinstance(e, (KafkaConnectionError, OSError, ConnectionError)):
logger.error("=" * 70)
logger.error("❌ CONNECTION ERROR")
logger.error("=" * 70)
logger.error(f" Error: {e}")
return None
else:
logger.error("=" * 70)
logger.error(f"❌ UNEXPECTED ERROR: {type(e).__name__}")
logger.error("=" * 70)
logger.error(f" Error: {e}")
return None
def on_success(metadata):
logger.info(f"✓ Message sent to topic '{metadata.topic}' partition {metadata.partition} at offset {metadata.offset}")
def on_error(e):
logger.error(f"❌ Error sending message: {e}")
error_str = str(e)
if "UNKNOWN_TOPIC_OR_PARTITION" in error_str:
logger.error(f" The topic '{TOPIC}' doesn't exist or you lack permissions")
elif "TOPIC_AUTHORIZATION_FAILED" in error_str:
logger.error("")
logger.error(" ⚠️ ACL PERMISSION ERROR - TOPIC_AUTHORIZATION_FAILED")
logger.error(f" User '{USERNAME}' doesn't have WRITE permission on topic '{TOPIC}'")
logger.error("")
logger.error(" Required permission: WRITE")
logger.error(" How to fix:")
logger.error(" 1. Log into Redpanda Console")
logger.error(" 2. Go to Security → ACLs")
logger.error(f" 3. Grant WRITE permission to user '{USERNAME}' on topic '{TOPIC}'")
logger.error("")
elif "AUTHORIZATION_FAILED" in error_str:
logger.error(f" ACL issue: User '{USERNAME}' lacks required permissions")
def main():
logger.info("="*70)
logger.info("Redpanda Producer - Smart Error Detection")
logger.info("="*70)
# Validate configuration
if not validate_config():
logger.error("\n⚠️ Fix configuration issues above")
sys.exit(1)
# Create producer
producer = create_producer()
if not producer:
logger.error("")
logger.error("="*70)
logger.error("⚠️ CONNECTION FAILED - See error details above")
logger.error("="*70)
sys.exit(1)
try:
hostname = str.encode(socket.gethostname())
logger.info("")
logger.info("="*70)
logger.info(f"Producing messages from: {socket.gethostname()}")
logger.info("="*70)
# Produce messages
success_count = 0
error_count = 0
for i in range(100):
msg = f"asynchronous message #{i}"
try:
future = producer.send(
TOPIC,
key=hostname,
value=str.encode(msg)
)
future.add_callback(on_success)
future.add_errback(on_error)
# Check result for first few messages to catch errors early
if i < 5:
try:
record_metadata = future.get(timeout=10)
success_count += 1
if i == 0:
logger.info(f"✓ First message sent successfully!")
except Exception as e:
error_count += 1
logger.error(f"Failed to send message #{i}")
on_error(e)
if "TOPIC_AUTHORIZATION_FAILED" in str(e) or "AUTHORIZATION_FAILED" in str(e):
logger.error("Stopping due to ACL error...")
break
else:
success_count += 1
except Exception as e:
error_count += 1
logger.error(f"Exception while sending message #{i}: {e}")
if error_count == 0:
logger.info("\nFlushing remaining messages...")
producer.flush()
logger.info("")
logger.info("="*70)
logger.info(f"✓ SUCCESS - All {success_count} messages sent")
logger.info("="*70)
else:
logger.error("")
logger.error("="*70)
logger.error(f"⚠️ COMPLETED WITH ERRORS - {success_count} sent, {error_count} failed")
logger.error("="*70)
except KeyboardInterrupt:
logger.warning("\n⚠️ Interrupted by user")
except Exception as e:
logger.error(f"❌ Unexpected error: {e}")
finally:
logger.info("\nClosing producer...")
producer.close()
logger.info("✓ Producer closed")
if __name__ == "__main__":
main()