Kafka Offset Management and Replication Mechanisms Explained

1. Message handling entry The entry point for processing Kafka messages on the broker side is the handle(request: RequestChannel.Request) method, which dispatches requests based on RequestKeys (e.g., ProduceKey, FetchKey, OffsetCommitKey, etc.) and includes error handling and response generation.

/**
   * Top-level method that handles all requests and multiplexes to the right api
   */
def handle(request: RequestChannel.Request) {
  try {
    trace("Handling request:%s from connection %s;securityProtocol:%s,principal:%s".format(request.requestObj, request.connectionId, request.securityProtocol, request.session.principal))
    request.requestId match {
      case RequestKeys.ProduceKey => handleProducerRequest(request)
      case RequestKeys.FetchKey => handleFetchRequest(request)
      // ... other cases ...
      case requestId => throw new KafkaException("Unknown api code " + requestId)
    }
  } catch {
    case e: Throwable =>
      if (request.requestObj != null)
        request.requestObj.handleError(e, requestChannel, request)
      else {
        val response = request.body.getErrorResponse(request.header.apiVersion, e)
        val respHeader = new ResponseHeader(request.header.correlationId)
        if (response == null)
          requestChannel.closeConnection(request.processor, request)
        else
          requestChannel.sendResponse(new Response(request, new ResponseSend(request.connectionId, respHeader, response)))
      }
      error("error when handling request %s".format(request.requestObj), e)
  } finally {
    request.apiLocalCompleteTimeMs = SystemTime.milliseconds
  }
}

2. In‑memory offset source Offsets are retrieved from the offsetsCache. When a consumer group requests offsets, the method checks if the group is local and either returns all cached offsets or specific partitions, otherwise it returns a not‑coordinator error.

/**
   * The most important guarantee that this API provides is that it should never return a stale offset.
   */
def getOffsets(group: String, topicPartitions: Seq[TopicAndPartition]): Map[TopicAndPartition, OffsetMetadataAndError] = {
  trace("Getting offsets %s for group %s.".format(topicPartitions, group))
  if (isGroupLocal(group)) {
    if (topicPartitions.isEmpty) {
      offsetsCache.filter(_._1.group == group).map { case (groupTopicPartition, offsetAndMetadata) =>
        (groupTopicPartition.topicPartition, OffsetMetadataAndError(offsetAndMetadata.offset, offsetAndMetadata.metadata, ErrorMapping.NoError))
      }.toMap
    } else {
      topicPartitions.map { topicAndPartition =>
        val groupTopicPartition = GroupTopicPartition(group, topicAndPartition)
        (groupTopicPartition.topicPartition, getOffset(groupTopicPartition))
      }.toMap
    }
  } else {
    debug("Could not fetch offsets for group %s (not offset coordinator).".format(group))
    topicPartitions.map { topicAndPartition =>
      val groupTopicPartition = GroupTopicPartition(group, topicAndPartition)
      (groupTopicPartition.topicPartition, OffsetMetadataAndError.NotCoordinatorForGroup)
    }.toMap
  }
}

3. Offset commit implementation When a consumer commits offsets, the broker treats the commit as a normal produce request to the internal __consumer_offsets topic. The commit is appended to the log, replicated to all ISR replicas, and only after successful replication (acks = -1) is the offset stored in the cache.

/**
   * Store offsets by appending it to the replicated log and then inserting to cache
   */
def prepareStoreOffsets(groupId: String, consumerId: String, generationId: Int,
                          offsetMetadata: immutable.Map[TopicAndPartition, OffsetAndMetadata],
                          responseCallback: immutable.Map[TopicAndPartition, Short] => Unit): DelayedStore = {
  val filteredOffsetMetadata = offsetMetadata.filter { case (_, offsetAndMetadata) =>
    validateOffsetMetadataLength(offsetAndMetadata.metadata)
  }
  val messages = filteredOffsetMetadata.map { case (topicAndPartition, offsetAndMetadata) =>
    new Message(key = GroupMetadataManager.offsetCommitKey(groupId, topicAndPartition.topic, topicAndPartition.partition),
                bytes = GroupMetadataManager.offsetCommitValue(offsetAndMetadata))
  }.toSeq
  // ... build ByteBufferMessageSet and send to log ...
  DelayedStore(offsetsAndMetadataMessageSet, putCacheCallback)
}

4. Offset fetch implementation Consumers request offsets at startup; the broker reads the offsets from the __consumer_offsets partitions. If the partition is still loading, the broker returns an OffsetLoadingCode response.

5. Kafka file storage layout Kafka stores data per topic in directories containing log segments, index files, and checkpoint files such as cleaner-offset-checkpoint, meta.properties, recovery-point-offset-checkpoint, and replication-offset-checkpoint. Each segment consists of an .log file and a corresponding .index file.

├── data0
│   ├── cleaner-offset-checkpoint
│   ├── client_mblogduration-35
│   │   ├── 00000000000004909731.index
│   │   ├── 00000000000004909731.log
│   │   ├── 00000000000005048975.index
│   │   ├── 00000000000005048975.log
│   ├── __consumer_offsets-33
│   │   ├── 00000000000000105157.index
│   │   └── 00000000000000105157.log
│   ├── meta.properties
│   ├── recovery-point-offset-checkpoint
│   └── replication-offset-checkpoint

6. Leader‑follower synchronization mechanism The article walks through a single‑partition, replication‑factor‑2 example, showing how a produce request updates the leader’s log (LEO) and how subsequent fetch requests from the follower update the remote LEO, compute the high‑watermark (HW) as min(leader LEO, follower remote LEO), and finally commit the HW after an additional fetch round.

Overall, the article explains how Kafka ensures consistency of offset data and message replication through careful handling of produce, fetch, commit, and load operations, highlighting potential pitfalls such as data loss or inconsistency when HW updates are delayed.