java_kafkasamples/docs/06 Consuming messages.md

# 06 Consuming messages with Kafka Consumers and Consumer Groups

Very similar to producer.

## Creating a consumer

- Properties:
    - `bootstrap.servers`
    - `key.serializer`
    - `value.serializer`
    - and others
    - => `KafkaProducer`
    - [Consumer config doc](http://kafka.apache.org/documentation.html#consumerconfigs)
- See [KafkaConsumerApp](../src/main/java/KafkaConsumerApp.java)

## (Un)Subscribing

- A single consumer can subscribe to any number of topics
- subscribe takes either a list of strings or a regex string,
    - each call replaces the previous subscription list
- They can also unsubscribe all topics at once.

### consumer.subscribe()

- automatic/dynamic partition assignment
    - that single consumer listens to every partition on every listed topic
    - entirely managed by the `KafkaConsumer`
        - will get newly added partitions to subscribed topics (`SubscriptionState`)
- now needs a valid `group.id` == `ConsumerConfig.GROUP_ID_CONFIG`

### consumer.assign()

- used to subscribe to explicit partitions
    - one or more partitions, regardless of topic
    - manual, self-administering mode
- more of an advanced case
- does not needs a `group.id`
- blocks instead of erroring if one of the partitions does not exist, in
  `updateFetchPositions.updateFetchPositions()` (see poll loop below).

## The poll loop

- primary function of the `KafkaConsumer`
- continuously poll the brokers for data
- single API for handling all Consumer <-> Broker interactions
    - a lot of interactions beyond message retrieval
- when `consumer.assign()` or `.subscribe()` is called, the contents of the topics and partitions collections in
  metadata are used to set fields within the
  `SubscriptionState consumer.subscriptions` field
    - this value is the source of truth for everything the consumer is assigned to
    - most consuming operations interact with it and the `ConsumerCoordinator consumer.coordinator` field
- when `consumer.poll()` is invoked, it uses properties (starting with the `bootstrap.servers`) to request metadata
  about the cluster.
- the `Fetcher<K, V> consumer.fetcher` performs several fetch-related operations between consumer and cluster
    - it delegates actual communication to the `ConsumerNetworkClient consumer.client`
    - that consumer, in addition to the actual requests, performs the heartbeats informing the cluster of the client
      health
    - the fetch also requests (via the CNC) the cluster metadata, initially then periodically afterwards
    - it obtains information about what topics/partitions are available from the
      `SubscriptionState`
- the `ConsumerCoordinator consumer.coordinator` uses the metadata to coordinate the consumer.
    - handles automatic/dynamic partition reassignment by notifying them to the SubscriptionState
    - committing offsets to the cluster so the cluster is aware of the state of subscriptions, for each topic and
      partition.
- the timeout applies to the time the CNC spends polling the cluster for messages to return. It does not apply to the
  initial setup operations (hence poll remaining stuck on nonexistent partitions during the initial update)
    - it is a minimum amount of time the cycle will take, not a maximum
- when the timeout expires, a batch of records are returned
    - they are parsed, deserialized, and stored by topic and partition internally in the Fetcher
    - once this is done, the fetcher returns this result for application processing.

The `poll()` process is a single-threaded operation:

- 1 poll loop per-consumer
- 1 thread per-consumer

This is for internal simplicity and to force parallel consumption to be performed otherwise.

## Processing messages

The `poll()` method returns a `ConsumerRecords`, a collection of `ConsumerRecord`:

```java
public class ConsumerRecords<K, V> implements Iterable<ConsumerRecord<K, V>> {
    /* ... */
}
```

- Since the `poll()` runs in a single thread, there is nothing running while the records are being processed
- The more topics and partitions a single process subscribes to, the more it has to do within that single polling loop,
  which can make the consumer slow

## Offsets and positions

Just because something is _read_ does not mean it is _committed_:

- There are different categories of offsets, representing the stage they're in:
    - a consumer needs to know what it has vs has not read
    - what it confirms it has read (and processed) is the _last committed offset_
    - this is whence consuming starts for a given partition, subject to offset reset
- Each partition is exclusive with regard to consumer offsets: for any topic, a
  consumer is tracking one offset per partition it is subscribed to
- Consumers read at the _current position_ towards the _log-end offset_.
- Offsets from the last committed to the current position are _uncommitted_.
- Two **optional** offset commit properties control these offsets:
    - `enable.auto.commit` lets Kafka decide when current position offsets are
      upgraded to full committed offsets. Defaults to `true`.
    - Since it has no knowledge of what work is actually performed, it can only
      use a delay for that. That delay is the `auto.commit.interval`, in msec,
      which defaults to 5000, which is a lot for high-throughput cases.

<blockquote>
The extent to which your system can be tolerant of eventual consistency
is detemined by its reliability.
</blockquote>

By default, consumers start reading from a new partition at the `latest` committed offset.

- Optional property:
    - `auto.offset.reset` can be `earliest`, `latest` (default), or `none` which
      throws an exception and lets code decide. See "rebalancing" below.

Offset choice is different depending on whether the topology has a single consumer,
or a ConsumerGroup.

- Kafka stores offsets in a special topic called `__consumer_offsets`, with 50 partitions.
- They are produced by the `ConsumerCoordinator`, which means a Consumer is also
  a Producer, but for just that topic. Once they have been committed, the coordinator
  updates the SubscriptionState accordingly, allowing the Fetch to know which
  offsets it should be retrieving.

## Manual committing

There are two methods to commit:

- `commitSync()`: used to achieve exact control of when the record is truly processed;
  when one doesn't want to process newer records until the older ones are committed.
    - It should be called after processing a **batch** of ConsumerRecord, not just a
       single record, which increases latency to no avail
    - It blocks until it receives a response from the cluster
    - It may fail, meaning a recovery process must be started. It retries automatically
      until it succeeds, or it receives an unrecoverable error.
    - The `retry.backoff.ms` == `RETRY_BACKOFF_MS_CONFIG` (default: 100) property defines the retry delay, similarly
      to the producer backoff (for good reason)
    - Using this manual committing is a tradeoff between throughput/performance (less)
      and control over consistency (more), as this adds some latency to the polling
      process.
- `commitAsync()`
    - does not retry automatically, because it can't know whether the previous
      commit succeeded or failed, which could lead to ordering and duplication issues
    - instead, it accepts a callback, triggered on receiving the commit response
      from the cluster.
    - better throughput because it doesn't block the polling process.
    - Don't use it without registering the callback and handling it as needed.

## Deciding on a commit strategy

- Consistency control:
    - need to know when something is "done" (define "done" first)
- Atomicity:
    - ability to treat consumption and processing as a single atomic operation
    - obtaining _exactly-once_ semantics instead of _at-least-once_

## Scaling out consumers

- Scaling a single-thread, single-consumer app to the bandwidth, number of topics
  and partitions of a full K cluster is not realistic
- The solution is to scale out consuming to more consumers, but they can't jusst
  consumer anything without synchronizing in some way
- This is the reason for Consumer groups: a collection of independent Consumer
  working as a team, i.e. declaring the same `group.id`.
    - This allows them to share the message consumption and processing load, with more parallelism.
    - It allows more redundancy: failure or limitations of a given consumer are
      automatically handled and balanced by K.
    - It offers more performance, with the ability to support a large backlog
- A Consumer group is create when individual consumers
    - with a commmon `group.id`...
    - invoke the `subscribe()` method...
    - and pass a common topics list.
- One of the brokers gets elected as the `GroupCoordinator` for that topic.
    - Its job is to monitor and maintain group membership.
    - It collaborates with the `ClusterCoordinator` and ZooKeeper to monitor and
      assign partitions within a topic to individual consumers in the group.
- As soon as a Consumer group is formed, each consumer is sending heartbeats,
    - configured by properties:
        - `heartbeat.interval.ms` == `HEARTBEAT_INTERVAL_MS_CONFIG` (default 3000 msec):
          the interval between heartbeat sends
        - `session.timeout.ms` == `SESSION_TIMEOUT_MS_CONFIG` (default 30000 msec)
    - the CG coordinator relies on these heartbeats to evalue whether the consumer
      is alive and able to participate in the group
    - if the coordinator does not receive heartbeat during a "total time" (?)
      larger than `session.timeout.ms`, it will consider the consumer failed and take
      corrective action, following its priority: ensuring that the purpose of the
      group (sharing the load of consuming those topics) is being met.
    - these corrections are _consumer rebalance_ operations, which is complex
        - remaining consumers now need to absorb the workload no longer handled
          by the failed consumer
        - they need to find up to where the failed consumer had worked (commit offset)
          for all partitions, and catch up without creating duplicates.
        - the ability to perform these rebalances is critical to cluster health.
        - example 1: the failed consumer handled messages but could not commit
          them; in that case the new consumer are likely to re-process them,
          possibly introducing duplicates
        - example 2: a new consumer joins the group
        - example 3: a new partition is added to the topic
- Applications are configured with a consumer group to handle their topics

## Rebalancing

The offset at which a new consumer in a group starts consuming is defined by
the `auto.offset.reset` == `AUTO_OFFSET_RESET_CONFIG` property.

If the rebalance was triggered at a point when a previous consumer had already
read but not yet committed some offset, the new consumer is likely to read it again.

The primary purpose of the Group Coordinator is to evenly ablance available consumers
to partitions.

- If possible, it will assign a 1:1 consumer/partition ratio.
- If there are more consumers than partitions, it will let the extra consumers idle,
  leading to over-provisioning
  - Apparently (to be confirmed), even if there are more partitions than consumers,
    it will not share a partition across multiple consumers
- If a new partition becomes available, or a consumer fails, or is added,
  the Group Coordinator initiates the rebalancing protocol, engaging each Consumer coordinator

## Consumer configuration: performance and efficiency

- `fetch.min.bytes` : minimum number of bytes returned from the poll. Analogous
  to the `batch.size` settings on the producer
- `max.fetch.wait.ms` maximum time to wait if that threshold has not yet been met.
  Analogous to the `linger.ms` setting on the producer
- `max.partition.fetch.bytes`: maximum number of bytes per partition returned in a
  `poll()` call, to avoid spending too much time on processing before calling
  `poll()` again.
- `max.poll.records` similar but in terms of records, not bytes, and not per partition

## Advanced topics not covered

- Consumer position control
    - `seek(TopicPartition, long)`: specify the offset from which to read on a
      given partition
    - `seekToBeginning(Collection<TopicPartition>)`: seek from the beginning of a group of topics and partitions
    - `seekToEnd(Collection<TopicPartition>)`: its opposite
- Flow control allows pausing/resuming subscription to some topics or partitions
    - `pause(TopicPartition)`
    - `resume(TopicPartition)`
    - mostly useful when a single consumer has to read multiple topics and partitions
- Rebalance listeners: get notified when rebalances occur in the cluster, to adjust
  in consumer code how to handle the offsets.