Problem
I needed to created topics in Amazon Web Services(AWS) Managed Streaming for Apache Kafka(MSK) and I wanted to list out the topics after they were created to verify.
Solution
This solution is written in python using the confluent-kafka package. It connects to the Kafka cluster and adds the new topics. Then it prints out all of the topics for verification
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from confluent_kafka.admin import AdminClient, NewTopic | |
| import os | |
| a = AdminClient( | |
| { | |
| 'bootstrap.servers': os.environ.get("AWS_KAFKA_BOOTSTRAP_SERVERS"), | |
| 'security.protocol': 'SASL_SSL', | |
| 'sasl.mechanisms': 'AWS_MSK_IAM', | |
| 'sasl.aws.access.key.id': os.environ.get("AWS_ACCESS_KEY_ID"), | |
| 'sasl.aws.secret.access.key': os.environ.get("AWS_SECRET_ACCESS_KEY"), | |
| 'sasl.aws.security.token': os.environ.get("AWS_SESSION_TOKEN"), | |
| 'sasl.aws.region': 'us-east-1', | |
| 'ssl.ca.location': '/etc/ssl/certs/ca-certificates.crt', | |
| } | |
| ) | |
| new_topics = [NewTopic(topic, num_partitions=3) for topic in ["topic1", "topic2"]] | |
| # Note: In a multi-cluster production scenario, it is more typical to use a replication_factor of 3 for durability. | |
| # Call create_topics to asynchronously create topics. A dict | |
| # of <topic,future> is returned. | |
| fs = a.create_topics(new_topics) | |
| # Wait for each operation to finish. | |
| for topic, f in fs.items(): | |
| try: | |
| f.result() # The result itself is None | |
| print("Topic {} created".format(topic)) | |
| except Exception as e: | |
| print("Failed to create topic {}: {}".format(topic, e)) | |
| topics = a.list_topics().topics | |
| print("There are currently {} topics. They are {}".format(len(topics), topics)) |
Cultivating Software 