aws_msk_iam_sasl_signer_kafka_python_producer_example.py

1import socket
2from kafka import KafkaProducer
3from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
4
5class MSKTokenProvider():
6    def __init__(self, region):
7        self.region = region
8
9    def token(self):
10        # The region must be the region where your MSK cluster is located.
11        # This will use the default AWS credential provider chain.
12        token, _ = MSKAuthTokenProvider.generate_auth_token(self.region)
13        return token
14
15# Replace these variables with your MSK cluster details
16region = "us-east-1"
17# The bootstrap servers can be obtained from the MSK console or using the AWS CLI.
18bootstrap_servers = ["b-1.example.kafka.us-east-1.amazonaws.com:9098"]
19
20tp = MSKTokenProvider(region)
21
22producer = KafkaProducer(
23    bootstrap_servers=bootstrap_servers,
24    security_protocol='SASL_SSL',
25    sasl_mechanism='MECHANISM_AWS_MSK_IAM',
26    sasl_oauth_token_provider=tp,
27    # It's recommended to set a client_id to identify your application.
28    client_id=socket.gethostname(),
29)
30
31# Example of sending a message
32producer.send('my-topic', b'Hello, MSK with IAM!')
33producer.flush()