Title: CS544 / 10-kafka
Tags: cs544-kafka

Q: What are two communication systems we've learned, for synchronous and asynchronous communication, respectively?
A: gRPC: sync
Kafka: async
gRPC uses protobufs for messages.  Kafka can use any format for messages (protobufs are a reasonable option).

Q: What are the three kinds of processes that work together in Kafka?
A: producers => brokers => consumers

Q: What is cron?
A: A program that can run other programs on a schedule

Q: What are some advantages of using a streaming platform like Kafka for ETL jobs, rather than running ETL jobs on a schedule (say with cron) to move data from system A to system B?
A: 1. data will be fresher if we're constantly streaming it from transactions processing systems to analytics systems
2. if you have many transactions processing systems and many analytics systems, you may need to write a custom job for every pair of systems.  Having a centralized log (like Kafka) means you just need to write one producer for each transactions processing system and one consumer for each analytics system

Q: Where is topics data stored in Kafka?
A: on brokers

Q: Kafka topics consist of one or more _____, each of which contains a series of ________
A: partitions
messages

Q: When a producer sends a message, what are 3 things it typically specifies?  And of the 3 things, what is optional?
A: topic, key, value
key is optional

Q: If a message doesn't have a key, how does a Kafka consumer decide what topic partition it should go in?
A: A "round robin" policy is used, meaning the producer just rotates between different partitions each time there is a new message.

Q: If a Kafka producer specifies a key for a message, how does it determine what partition the message should go to?
A: partition = hash(key) % partition_count
Note that this is pluggable -- you could define a different policy to override the default if you wanted to.

Q: How does a Kafka consumer request a batch of messages from the brokers, with a timeout of N milliseconds?
A: batch = consumer.poll(N)

Q: What does a Kafka batch contain?
A: messages, divided by topic partition

Q: If Kafka message A was produced before message B, when can we guarantee that consumers will see A before B?
Assume we don't know the partition count, but that whatever it is won't change.
A: We can only guarantee this when A and B have the same topic AND key.

Q: Kafka consumers typically read messages at ever increasing offsets in each of the tracked topic partitions, but if consumers want to move to a different offset, they can call this:
consumer.?????(partition, offset)
A: consumer.seek(partition, offset)

Q: Why might you want to have different Kafka consumer groups?
A: There are cases you might want the same messages to be seen by multiple different applications.  You can achieve this by using different consumer groups for each application.
Example:
c = KafkaConsumer("clicks", group_id="g1", ...)

Q: Why might you want multiple Kafka consumers in the same consumer group?
A: If the consumer group is subscribed to a high-volume topic (meaning the topic is getting new messages at a high rate), it will help the system scale if different consumers running on different machines can be responsible for different partitions of that topic.

Q: What is the difference between log "rollover" and "deletion" in Kafka?  What does "retention" mean?
A: At any given time, new data for a topic partition is being written to an "active" file.  Due to limits on file size/age, we might want to switch to writing to a new file, while still keeping the old file around in a finalized state.  This is "rollover".
"deletion" refers to deleting non-active files that are too old, or to meet some other criteria specified by a "retention" policy.

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)

Q: When is a Kafka message considered "committed"?
A: When it is written to ALL in-sync replicas.

Q: In Kafka, both consumers and follower replicas fetch data from the leader replica.  But how do the fetches differ?
A: Consumers fetch only "committed" messages.
Followers fetch both committed and uncomitted messages.
This is because we only want consumers to see committed messages.  But the way messages become committed is by being received by more replicas, so the follower replicas need to fetch uncommitted data.

Q: If min.insync.replicas is X and the number of currently in-sync replicas is Y, why doesn't Kafka consider a message committed as soon as it is written to at least X replicas?  (assume X<Y in this question)
A: Hypothetical scenario showing why sending an ACK in this scenario could result in loss of committed data:
Let's say X=2 and Y=3.
1. write occurs on two replicas (one "in-sync" replica doesn't have the message)
2. leader returns an "ack" indicating message is committed at offset 100
3. leader dies
4. Kafka selects a new leader from those that are in-sync -- let's say it selects the one that didn't receive the message
5. the new leader is the authority on what messages are at each offset, and might write a different messages at offset 100.  This will mean a previously committed message will have changed
That's why a Kafka leader WILL NOT return an ack in step 2.  It will wait until the message is on the third replica too.

Q: In order to be considered "in sync", does a follower replica in Kafka need to have ALL the messages the leader has seen?
A: No, it is allowed to lag a bit.  This is configurable -- for example, you might define "in sync" to mean the follower has retrieved a batch reading the end of the topic partition in the last N seconds.

Q: How do "elections" in distributed systems usually differ from a regular democratic election humans might participate in?
A: In a human election, we usually have a couple goals:
1. represent the will of the people
2. have a well defined process so it is clear who the leader is (it is problematic if there is disagreement about who the leader is)
In distributed systems, we usually just care about (2).  So in Kafka, for example, "election" of a replica leader is really just a special "controller" broker making a decision without input from other brokers.

Q: What is an example of a "write anomoly" in Kafka?
A: A message was acknowledged as successfully written, but it was not actually readable later.

Q: What is an example of a "read anomoly" in Kafka?
A: A message was read, but when you try to read it again later, it has dissapeared.

Q: Producers can be created like this: KafkaProducer(..., acks=????)
What are the options, and what do they each do?
A: 0: don't wait for the leader to send an ack
1: ack means leader has written message to its own log
"all": message has been written to all in-sync replicas

Q: How might we end up with "at-least-once" semantics?
A: This happens when a sender keeps retrying until it gets an acknowledgement of success.  If the ack is received on the Nth time, then for the first (N-1) times, the message may or may not have gone through.  This means the message will have been sent between 1 and N times.

Q: How might you end up with "at-most-once" semantics?
A: You try to send a message, and you never retry sending, even if it isn't acknowledged.  If not acknowledged, there's still a chance the message made it through, and the failure happened before the ack could be sent.  So the messages will have been received either 0 or 1 times.

Q: What does it mean for an operation to be idempotent?
A: the effect of performing the operation multiple times is equivalent to the effect of performing the operation a single time

Q: How might we make an operation idempotent, even if it not naturally so?
A: Associate the operation with a unique ID, and whatever system is receiving the operation requests is responsible for remembering previously seen IDs and supressing/ignoring duplicates.

Q: What is the problem with "auto commit" for Kafka consumers?
A: Auto commit happens on some timer, and it's possible the consumer created some output (based on messages in a batch), then crashed before commiting the offset after those messages.  When a new consumer takes over, it will start from slightly old offsets, and re-process messages that have already influenced the output.  This could result in "overcounting" or other incorrect output.

Q: If we have Kafka consumers (a) processing messages from a Kafka topic and (b) writing output based on that processing, how can we guarantee each message is considered exactly once in the output, even if the consumer crashes and restarts?
A: Whenever the consumer writes output, we want to simultaneously (that is, "atomically") write offsets to indicate what messages were used to create that output.  For example, if the consumer is writing to a MySQL database, it might do two INSERTs as part of the same transaction (one INSERT for the actual output, and another INSERT to record the corresponding offsets somewhere).