create_topics.py

gistfile1.txt

import kafka.conn
from kafka import protocol
import collections

ip, port, afi = kafka.conn.get_ip_port_afi('localhost:9092')
bc = kafka.conn.BrokerConnection(ip, port, afi)
bc.check_version()

bc.send(protocol.admin.ListGroupsRequest[0]())
bc.recv()

Topic = collections.namedtuple('Topic', 'topic num_partitions replication_factor replica_assignment configs')

req = protocol.admin.CreateTopicsRequest[0](
    create_topic_requests=[Topic(topic='example', num_partitions=12, replication_factor=-1,replica_assignment=[],configs=[])],
    timeout=1000
)
fut = bc.send(req)
bc.recv()