Title: 544
Tags: lec32

Q: If your want to connect to a Kafka broker "localhost:9092", how would you do it in these cases?
from kafka import KafkaAdminClient, KafkaProducer, KafkaConsumer
admin = KafkaAdminClient(????)
producer = KafkaProducer(????)
consumer = KafkaConsumer(????)
A: This can be pasted in where each "????" is:
bootstrap_servers=["localhost:9092"]

Q: How can you create a Kafka topic "T" with P partitions and R replicas?
A: admin.create_topics([NewTopic("T", num_partitions=P, replication_factor=R)])

Q: How can a Kafka producer send the string "hello" to topic "greetings"?
A: producer.send("greetings", value=bytes("hello", "utf-8"))
Values must always be represented as bytes!

Q: How does a Kafka consumer check for a new batch of messages, with timeout of 1 second?
A: batch = consumer.poll(1000)

Q: How can you see the topic partitions a Kafka consumer is responsible for?
A: consumer.assignment()

Q: If you want a Kafka consumer to go back the start of each topic partition, how can you do that?
A: consumer.seek_to_beginning()

Q: Assignment of topic partitions to a Kafka consumer can happen automatically or manually.  What does the code for each look like?
A: automatic:
consumer.subscribe(["some_topic"])  # Kafka decides which partitions of some_topic
manual:
consumer.assign([TopicPartition("some_topic", 3)])    # this consumer will only read from partition 3 of some_topic

Q: If tp is a topic partition, how can you check the next offset a Kafka consumer will read at for that partition?
A: offset = consumer.position(tp)