Kafka Fundamentals

high level overview

what is kafka?

Kafka is an event streaming platform, combining three key capabilities:

1. To publish (write) and subscribe to (read) streams of events
2. To store streams of events durably and reliably for as long as you want
3. To process streams of events as they occur or retrospectively

it's a log: an ordered, persistent sequence of events

events are organized in topics

topics are split into partitions

producers write to topic partitions

consumers read from topic partitions

that's it?

kafka cluster

replicated logs — leaders and replicas

replicated logs — broker (default) config

replicated logs — topic creation/config

$ bin/kafka-topics.sh --create --topic my-topic \
--partitions 5 \
--replication-factor 3 \
--config retention.ms=864000000

replicated logs — in-sync replicas (ISR)

leader election

• Is triggered when a leader replica fails, or when a new replica is added to a partition
• Controlled by the Kafka broker that acts as the (*active) controller for the cluster

leader election - related config

that's the cluster (?)

Producers

Producer overview

connect to broker(s) and cache metadata

• bootstrap.servers – list of host/port pairs for establishing the initial connection to the cluster. Is used to discover the full set of servers/brokers
• At least 2 are recommended for fault-tolerance
• The bootstrap server returns list of all brokers in the cluster and all metadata like topics, partitions, partition leader, replication factor etc.
• metadata.max.age.ms – update metadata even if there's no leadership change happening due to new brokers or topics / partitions (default: 5 min)

producer record

serialize key/value

• key.serializer – Serializer class for key, e.g. ...StringSerializer
• value.serializer – Serializer class for value, e.g. ...ByteArraySerializer
• Not covered here: Confluent Schema Registry

select topic partition

Messages with the same key are written to the same partition – ordering is provided

buffer/accumulate messages

The send() method is asynchronous, it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

buffer/accumulate messages (config)

• batch.size – batch size in bytes, i.e. the upper bound of the batch size to be sent (default: 16 kB)
• linger.ms – time to give for batches to fill, to make batching more likely even under moderate load (default: 0)
• buffer.memory – the total amount of memory available for buffering. If records are sent faster than they can be transmitted to the server this buffer space will be exhausted (default: 32 MB)
• max.block.ms – max time to block send() calls when the buffer space is exhausted, after which a TimeoutException is thrown (default: 1 min)

send message batches

Batches are grouped by broker and sent if either batch.size or linger.ms is reached.

send message batches in parallel

max.in.flight.requests.per.connection controls the number of messages to be sent without any acknowledgment (default: 5)

message batch send failures

With max.in.flight.requests.per.connection > 1 there could be gaps if a request fails

handling send failures (retries)

• retries – number of times to resend a failed request (default: Integer.MAX_VALUE)
• request.timeout.ms – max time the client will wait for the response (default: 30 sec)
• delivery.timeout.ms – max time for send() to complete (default: 2 min)
• Note: with max.in.flight.requests.per.connection > 1 and retries > 0 message reordering could occur

handling send failures (retries)

To allow retries and prevent message reordering set max.in.flight.requests.per.connection = 1

send failure - duplicates

If the response gets lost, enabled retries could lead to duplicate messages
(at-least-once delivery semantics)

idempotent producer

• enable.idempotence – prevents message duplication and reordering for a single producer (!), to achieve exactly-once delivery semantics (default: true for Kafka 3.x)
• Requires max.in.flight.requests.per.connection <= 5 (with message ordering preserved), retries > 0 and acks='all'
• If conflicting configurations are set and idempotence is not explicitly enabled, it is disabled. If explicitly enabled and conflicting configs are set, a ConfigException is thrown
• Application level retries e.g. due to a app/producer crash could still lead to duplicates

message durability / consistency

• acks – number of acknowledgments the producer requires the leader to have received before considering a request complete (default: all for Kafka 3.x)
• acks=0 – do not wait for any acknowledgment
• acks=1 – leader writes the record and responds without awaiting acks from followers
• acks=all – leader waits for min.insync.replicas to acknowledge. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive

that's the producer (?)

Consumers

Consumer overview

• connect to broker(s) and cache metadata
• subscribe to topic(s)
• join consumer group
• continuously: poll records from broker(s) and commit offsets

							
								Properties props = new Properties();
								props.setProperty("bootstrap.servers", "localhost:9092");
								props.setProperty("group.id", "test");
								KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
								consumer.subscribe(Arrays.asList("foo", "bar"));
								while (true) {
									ConsumerRecords<String, String> records = consumer.poll(100);
									for (ConsumerRecord<String, String> record : records) {
										process(record);
									}
								}
							
						

consumer offsets

• committed offset: where to resume from in the event of a restart
• auto.offset.reset – how to behave when no offsets have been committed, or when a committed offset is no longer valid (earliest/latest/none, default: latest)

consumer offset management

• The default enable.auto.commit=true means that offsets are committed automatically with auto.commit.interval.ms (default: 5 sec)
  
  ⚠️ For at-least-once delivery you must consume all data returned from poll() before any subsequent calls, or before closing the consumer (otherwise, the committed offset could get ahead of the consumed position)
• It's also possible to "manually" commitSync or commitAsync an offset
  • commitSync: the consumer is blocked until the commit request returns successfully, including retries
  • commitAsync: send the commit request and return immediately to process the next message or batch of messages

consumer offset storage

• Kafka stores offsets in a compacted topic called __consumer_offsets
• offsets.retention.minutes – retention period (default: 7 days)
• Alternative: when processing to an external system, the consumer's offset can be stored together with the output (atomically), to be used with seek(TopicPartition, long) on restart

consumer offset lags

• With a growing consumer lag eventually messages could get lost
• Increasing the log retention time could be a temporary workaround
• Adding consumers (and/or partitions) could increase throughput
• Increasing fetched batches could increase throughput (with the cost of latency)
  • fetch.max.wait.ms – sets a maximum threshold for time-based batching (default: 500)
  • fetch.min.bytes – sets a minimum threshold for size-based batching (default: 1)

consumer groups

Ensure, that a single topic partition is processed by 1 consumer of the group.
A consumer group is identified by a group id.

consumer group rebalancing

The group coordinator is responsible for managing
a) group members and b) their partition assignments.

consumer group rebalancing

• In the past rebalancing was "stop the world", with Kafka 2.4 incremental cooperative rebalancing was introduced to reduce consumption downtime
• partition.assignment.strategy – a list of class names, ordered by preference, of supported partition assignment strategies to distribute partition ownership (default: [RangeAssignor, CooperativeStickyAssignor])
• I.e. the default config (Kafka 3.3) uses RangeAssignor, but allows upgrading to CooperativeStickyAssignor with just a single rolling bounce (removing RangeAssignor from the list) (see also KIP-726 to change the default)
• For newer clusters CooperativeStickyAssignor should be preferred to get cooperative rebalancing
• A nice explanation / visualization of different strategies can be found in this blog post

group management: consumer liveness

• Each member in the group must send heartbeats to the coordinator to indicate its liveness. If no heartbeats are received before the expiration of the configured session timeout, the coordinator will remove this client from the group and initiate a rebalance.
• session.timeout.ms – max time that can pass without a heartbeat (default: 45 sec, got increased from 10 sec for 3.0, see KIP-735)
  ⚠️ Make sure to close() the consumer on shutdown to keep processing latency low!
• heartbeat.interval.ms – time between heartbeats sent to the consumer coordinator; must be < session.timeout.ms, recommended to be < 1/3 of it; also used for discovering that a rebalance is in progress (default: 3 sec)
• max.poll.interval.ms – max delay between invocations of poll(). If exceeded, the consumer is considered failed and the group will rebalance (default: 5 min)
• max.poll.records – max number of records that poll() will return; because the consumer will only join a rebalance inside poll(), reducing this config reduces the delay of a group rebalance (default: 500)

that's the consumer (?)

Questions?

Further Reading

A collection of links that could be of interest

• Fine-tune Kafka performance with the Kafka optimization theorem
• How to Choose the Number of Topics/Partitions in a Kafka Cluster?
• Optimizing Kafka producers
• Optimizing Kafka consumers