Page Comparison

Kafka Consumer : Required Properties

...

// Set the topic subscription or partition assignments
myConsumer.subscribe(topics);
myConsumer.assign(partitions);

try{
  while(true){
    ConsumerRecords<String, String> records = myConsumer.poll(100);
    //add business logic if needed in a loop for
  }
}finally{
  myConsumer.close();
}

Kafka Consumer : Polling

The response from the brokers is used to instantiate its internal metadata object, which will keep up to date, while the poll method runs, getting periodic updates from the cluster when cluster details change. With metadata available, other elements become more involved : ConsumerCoordinator has 2 tasks.

First, ConsumerCoordinator must be aware of partition reassignment and notification of assignment changes to the subscription state object ; Second, ConsumerCoordinator for committing offsets to the cluster.

The tasks will cause the confirmation of which will cause the update of the subscription state, so it can always be aware of the status of the topics and partitions.

...

Kafka Consumer and Offset

The offset is the critical value that enables consumer to operate independently by representing the last read position the consumer has read from a partition within a topic.

3 categories of offsets:

• last committed offset : last record that consumer has confirmed to have processed.

• current position : consumer tracks its current position.

• log-end offset : this position advances as consumer advances in the log towards the last record in the partition which is know as the log end offset.

...

Mind the Offset Gap or un-committed offsets : between offset 1 et offset 3

...

The coordinator is the responsible object for communicating to the cluster and for ensuring the committed offsets are produced into the topic.

Kafka Consumer and Post-Processing as commitSync

To handle the new record to process, we need to make sure that the ones we’ve currently processed are committed.

try{
  for(...){ //Processing batches of records...  }
  //Commit when we know we're done, after the batch is processed
  myConsumer.commitSync();
}catch(CommitFailedException){
  log.error("...");
}

Kafka Consumer and Post-Processing as commitAsync

To handle the new record to process, we need to control when to consider the message as truly processed.

try{
  for(...){ //Processing batches of records...  }
  //Commit when we know we're done, after the batch is processed
  myConsumer.commitAsync();//not recommended
  //recommended
  myConsumer.commitAsync(new OffsetCommitCallback(){
       public void onComplete(...,...,...){ //do something...}
  });
}catch(CommitFailedException){
  log.error("...");
}

Committing Offsets

...