Kafka Producer (Lambda) not sending messages to the Kafka broker (MSK)

Question

Hey everyone,
To give everyone some context - I have a lambda function that acts as a Kafka producer. On an S3 upload, the producer should send a message ( {"bucket", "key"} ) to the broker but for some reason its not sending any messages.
I am using serverless MSK in a private subnet so hard to access from the outside and check the logs.
Here's the lambda code:

```
from kafka import KafkaProducer
from kafka.sasl.oauth import AbstractTokenProvider
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
import os, json, time, socket, sys
from kafka.admin import KafkaAdminClient, NewTopic
from kafka import TopicPartition

class MSKTokenProvider(AbstractTokenProvider):
    def token(self):
        token, _ = MSKAuthTokenProvider.generate_auth_token('ca-central-1')
        return token

def create_topic(bootstrap_servers, topic_name, token_provider, num_partitions=1, replication_factor=2):
    print(f"Creating topic: {topic_name}")
    admin_client = None
    try:
        admin_client = KafkaAdminClient(
            bootstrap_servers=[bootstrap_servers],
            security_protocol='SASL_SSL',
            sasl_mechanism='OAUTHBEARER',
            sasl_oauth_token_provider=token_provider
        )

existing_topics = admin_client.list_topics()
        if topic_name in existing_topics:
            print(f"Topic: {topic_name} already exists")
            return

new_topic = NewTopic(name=topic_name, num_partitions=num_partitions, replication_factor=replication_factor)
        admin_client.create_topics(new_topics=[new_topic], validate_only=False)
        print(f"Topic '{topic_name}' created successfully.")
        
    except Exception as e:
        print(f"Error creating topic '{topic_name}': {e}")
    finally:
        if admin_client:
            admin_client.close()

def lambda_handler(event, context):
    token_provider = MSKTokenProvider()
    bootstrap_servers = os.getenv('BOOTSTRAP_SERVERS')
    topic = os.getenv('TOPIC_IN')
    group_id = os.getenv('GROUP_ID')
   
    producer = KafkaProducer(
        bootstrap_servers=[bootstrap_servers],
        security_protocol='SASL_SSL',
        sasl_mechanism='OAUTHBEARER',
        sasl_oauth_token_provider=token_provider,
        acks="1",
        retries=0,
        max_block_ms=5000,   
        request_timeout_ms=10000
    )

create_topic(bootstrap_servers, topic, token_provider)
    
    print(f"Kafka producer connected: {producer}")
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    print(f"Bucket name: {bucket_name}")
    print(f"Key: {key}")
    producer.send(topic, json.dumps({'bucket': bucket_name, 'key': key}).encode('utf-8'))
    
    producer.flush()
    print(f"Message sent to Kafka topic: {topic}")
    return {
        'statusCode': 200,
        'key': key
    }
```

From the lambda logs I can only see the logs until "Key: ..." and after neither is it throwing any error nor continuing the execution.

When I did some debugging like putting a log after the send statement even that was being executed suggesting that producer.flush() is blocking the execution and not allowing the script to finish running.

On the networking side and IAM side everything is configured correctly like MSK allows inbound traffic on port 9098 from Lambda's SG, whereas Lambda's SG allows egress to MSK's SG.

I have been scratching my head since the past 2 days - would really appreicate any help as to why the flush is behaving that way and particularly any solution to resolve this issue!

Also, not seeing any logs makes it really hard to debug!

Manoj Kengudelu · Answer

My few cents:

1. Please Check CloudWatch logs for specific error messages
2. If MSK is in a VPC, ensure Lambda is in the same VPC or has VPC connectivity. Basically verify your MSK cluster is accessible from the Lambda function's network location.
3. Make sure Lambda execution role has MSK permissions: ex: 
kafka-cluster:Connect
kafka-cluster:AlterCluster
kafka-cluster:DescribeCluster
kafka-cluster:WriteData
kafka-cluster:CreateTopic
kafka-cluster:DescribeTopic
4. Make sure to close the Producer.

```

try:
    # producer code
finally:
    producer.close()
```
5. You're creating the topic and immediately trying to send to it. Add a delay or retry logic

```
create_topic(bootstrap_servers, topic, token_provider)
time.sleep(2)  # Give topic time to be ready
```

re:Post Agent · Answer

Based on your description, it appears that your Lambda function is hanging during the `producer.flush()` operation when trying to send messages to your MSK cluster. This is likely due to one of several common issues with Kafka client connectivity.

First, let's address why `producer.flush()` is blocking execution. This method waits until all outstanding messages have been acknowledged by the Kafka broker. If the broker isn't responding or there's a connectivity issue, this call will block until it times out.

Here are some potential causes and solutions:

1. **Client Failover Issues**: When a Kafka client can't reach a broker or the broker isn't the leader for a partition, it will try to update its metadata and retry. This typically resolves within 1-3 seconds but can take longer if there are configuration issues. Your current setup with `retries=0` means it won't retry failed sends, which could be problematic.

2. **Authentication Problems**: Since you're using IAM authentication with SASL_OAUTHBEARER, verify that your Lambda function's execution role has the necessary permissions to interact with the MSK cluster.

3. **Network Configuration**: Even though you mentioned the security groups are configured correctly, double-check that the Lambda function can reach the MSK cluster on port 9098. Since your MSK cluster is in a private subnet, ensure that the Lambda function is also in the same VPC and has proper subnet configurations.

4. **Client Configuration**: Your current configuration has some potentially problematic settings:
- `retries=0`: This disables retries completely
- `max_block_ms=5000`: This is quite low and might cause the producer to give up too quickly
- `request_timeout_ms=10000`: This might not be enough time for the request to complete

Try modifying your producer configuration:
```python
producer = KafkaProducer(
bootstrap_servers=[bootstrap_servers],
security_protocol='SASL_SSL',
sasl_mechanism='OAUTHBEARER',
sasl_oauth_token_provider=token_provider,
acks="1",
retries=5,  # Allow some retries
retry_backoff_ms=100,  # Time between retries
max_block_ms=60000,  # Increase blocking time
request_timeout_ms=30000  # Increase request timeout
)
```

Also, consider adding error handling around your `send()` and `flush()` operations:
```python
try:
future = producer.send(topic, json.dumps({'bucket': bucket_name, 'key': key}).encode('utf-8'))
# Get the result with a timeout to avoid blocking indefinitely
record_metadata = future.get(timeout=30)
print(f"Message sent to partition {record_metadata.partition} at offset {record_metadata.offset}")

producer.flush(timeout=30)  # Add a timeout to flush
print(f"Message successfully flushed to Kafka topic: {topic}")
except Exception as e:
print(f"Error sending message to Kafka: {e}")
finally:
producer.close(timeout=5)  # Always close the producer
```

This approach will give you more visibility into what's happening and prevent indefinite blocking.

If you're still experiencing issues, you might want to check if your topic exists and has the correct configuration by examining the output of your `create_topic` function. Also, consider temporarily increasing the Lambda function timeout to give more time for debugging.  
**Sources**  
[Broker offline and client failover - Amazon Managed Streaming for Apache Kafka](https://docs.aws.amazon.com/msk/latest/developerguide/troubleshooting-offlinebroker-clientfailover.html)  
[AWS MSK is not able to load balance records to all the consumers in a consumer group | AWS re:Post](https://repost.aws/questions/QU4h24dFpfQ0Wx0JhPzgm_Ww/aws-msk-is-not-able-to-load-balance-records-to-all-the-consumers-in-a-consumer-group)

Kafka Producer (Lambda) not sending messages to the Kafka broker (MSK)

Add your answer

Relevant content