create_kafka_topic.py

from kafka.admin import KafkaAdminClient, NewTopic
import json

config_file = "kafka_topics_config.json"
with open(config_file, "r") as f:
    config = json.load(f)

client_id = config["client_id"]
bootstrap_servers = config["bootstrap_servers"]

admin_client = KafkaAdminClient(
    bootstrap_servers=bootstrap_servers, 
    client_id=client_id
)
num_partitions = 1
replication_factor = 1

topic_list = []
for topic in config["topic"]:
    name = topic["name"]
    num_partitions = topic["num_partitions"]
    replication_factor = topic["replication_factor"]
    topic_list.append(NewTopic(name=name, num_partitions=num_partitions, replication_factor=replication_factor))

admin_client.create_topics(new_topics=topic_list, validate_only=False)
print(f"all topics from config file {config_file} created")

docker-compose-kafka.yaml

version: "3.6"

services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2182:2181"
    networks:
      - outside
  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: localhost
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    networks:
      - outside

networks:
  outside:
    external:
      name: tutorial

kafka_topics_config.json

{
    "client_id": "admin",
    "bootstrap_servers": "localhost:9092",
    "topic": [
        {
            "name": "tutorial_topic",
            "num_partitions": 1,
            "replication_factor": 1
        }
    ]
}