Understanding Replication

1. Understanding Replication Concept

Aspect	Description	Details
Replication Factor	Number of copies of a partition	Default is 3; can be configured per topic
Leader Replica	Partition leader that handles all read/write requests	Selected by the controller; can change if leader fails
Follower Replicas	Copies of the leader that replicate data	Receive data from leader and maintain consistency
In-Sync Replicas (ISR)	Replicas that are up-to-date with the leader	Only ISR replicas can become leaders in case of failure
Unclean Leader Election	Allows non-in-sync replicas to become leaders	Enabled by setting `unclean.leader.election.enable=true`

Example: Configuring Replication Factor

kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

Note: Increasing replication factor improves fault tolerance but also increases storage overhead.

Warning: Setting unclean.leader.election.enable=true can lead to data loss if not carefully managed.

2. Understanding Leader and Followers

Role	Description	Responsibilities
Leader	The broker that handles all read and write requests for a partition.	Processes all client requests Manages replication consistency Updates the partition's metadata
Follower	Brokers that replicate the partition from the leader.	Replicates data from the leader Stays in sync with the leader Can become leader if the current leader fails

Example: Partition Leadership Change

kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092

Note: If the leader broker becomes unavailable, one of the followers will be elected as the new leader to maintain availability.

3. Configuring Replication Factor

Property	Default Value	Description
replication.factor	`1`	Controls the number of replicas for each partition
min.insync.replicas	`1`	Minimum number of replicas that must acknowledge a write before it is considered successful

Example: Setting Replication Factor at Topic Level

kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092

Example: Configuring min.insync.replicas

kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --config min.insync.replicas=2 --bootstrap-server localhost:9092

Note: The replication factor must be less than or equal to the number of brokers in the cluster.

4. Understanding In-Sync Replicas

Property	Description	Default Value	Notes
min.insync.replicas	Minimum number of replicas that must acknowledge a write for the write to be considered successful	2	Controls durability and availability trade-offs
unclean.leader.election.enable	Allows replicas that are not in-sync to become leaders (risk of data loss)	false	Set to true only if strict availability is required over consistency

Example: Configuring In-Sync Replica Settings

{"min.insync.replicas": 3,"unclean.leader.election.enable": false}

Note: A broker is considered in-sync if it is alive and has caught up to the latest messages in the log.

Warning: Setting min.insync.replicas too high can lead to write failures during broker outages.

5. Configuring Replica Lag

Property	Default Value	Description
replica.lag.time.max.ms	`30000`	Maximum time a replica may lag behind the leader before being considered out of sync.
replica.lag.min.config.ms	`5000`	Minimum time a replica must be lagging before it is considered for reassignment.
replica.lag.max.timeout.ms	`60000`	Maximum timeout for replica lag checks before triggering a reassignment.

Example: Configuring Replica Lag in server.properties

replica.lag.time.max.ms=60000
replica.lag.min.config.ms=10000
replica.lag.max.timeout.ms=120000

Note: Adjusting these values can impact the consistency and availability of your Kafka cluster. Lower values may cause frequent replica reassignments during temporary network issues.

6. Understanding Leader Election

Property	Description	Default Value	Notes
leader.imbalance.check.interval.ms	Interval in milliseconds to check for leader imbalance	`30000`	Used to detect and correct leader distribution imbalances
leader.imbalance.per.broker.percentage	Maximum percentage of partitions that can be unbalanced per broker	`10`	Threshold for triggering leader reassignment
unclean.leader.election.enable	Allow follower brokers to become leaders even if they are not in-sync	`false`	Setting to `true` may cause data loss

Example: Configuring Unclean Leader Election

{
  "unclean.leader.election.enable": true,
  "leader.imbalance.per.broker.percentage": 15
}

Note: Enabling unclean leader election can lead to data inconsistency if the new leader is not fully up-to-date with the previous leader.

Leader Election Process

Kafka elects a new leader when the current leader fails or becomes unavailable
Leaders are chosen from in-sync replicas (ISRs)
If unclean.leader.election.enable is set to true, out-of-sync replicas may be elected as leaders
Leader election is handled automatically by the controller broker

7. Understanding Replica High Watermark

Property	Description	Notes
high.water.mark	Indicates the offset of the last message that has been successfully replicated to all in-sync replicas (ISR).	Used by consumers to determine the latest offset they can safely read from.
log.end.offset	Represents the highest offset in the log, including unreplicated messages.	May be higher than high watermark if some messages are not yet replicated to all ISR members.

Example: Replica High Watermark in Practice

// Simulated Kafka consumer reading from a topic
const topic = "my-topic";
const consumer = new KafkaConsumer({
  bootstrapServers: ["localhost:9092"],
  groupId: "my-group"
});

// Consumer reads up to the high watermark offset
const messages = await consumer.fetch({
  topic,
  from: 0,
  to: highWaterMarkOffset
});

console.log("Consumed messages up to:", highWaterMarkOffset);

Note: The high watermark is updated only after a message has been successfully replicated to all in-sync replicas (ISR). If a broker is removed from ISR, the high watermark may lag behind the log end offset until replication completes.

8. Understanding Log End Offset

Property	Description	Value
log.end.offset	Represents the offset of the last message in a partition's log	`12345`
partition.offsets	Array of offsets for each partition in a topic	`[100, 200, 300]`

Example: Fetching Log End Offset

const kafka = new Kafka({ brokers: ['localhost:9092'] });
const admin = kafka.admin();
await admin.connect();

const topic = 'my-topic';
const partitions = await admin.fetchTopicMetadata({ topics: [topic] });

console.log('Log End Offsets:', partitions.topics[0].partitions.map(p => p.logEndOffset));

Note: The log end offset is crucial for consumers to know the latest available message in a partition.

9. Configuring Preferred Leader

Property	Default Value	Description
leader.imbalance.per.broker.percentage	10	Maximum imbalance percentage allowed per broker before triggering leader rebalancing
leader.imbalance.check.interval.ms	300000	How often to check for leader imbalance (5 minutes)
unclean.leader.election.enable	false	Whether to allow elections where a non-sync replica is elected as leader

Example: Preferred Leader Configuration

{
  "leader.imbalance.per.broker.percentage": 15,
  "leader.imbalance.check.interval.ms": 600000,
  "unclean.leader.election.enable": false
}

Note: Setting unclean.leader.election.enable to true can lead to data loss and should be used with caution.

10. Understanding Replica Fetcher

Property	Description	Default Value
replica.lag.time.max.ms	Maximum time a replica can lag behind the leader before being considered out of sync.	`30000`
replica.socket.timeout.ms	Socket timeout for replica fetch requests.	`30000`
replica.socket.receive.buffer.bytes	Size of the socket receive buffer for replica fetch requests.	`65536`
replica.fetch.max.bytes	Maximum number of bytes a replica can fetch in one request.	`1048576`
replica.fetch.min.bytes	Minimum number of bytes a replica must fetch in one request.	`1`
replica.fetch.wait.max.ms	Maximum time a replica fetch request can wait for data before returning.	`500`

Example: Configuring Replica Fetcher Settings

{
  "replica.lag.time.max.ms": 60000,
  "replica.socket.timeout.ms": 30000,
  "replica.fetch.max.bytes": 2097152
}

Note: Adjusting replica.lag.time.max.ms can impact the consistency and availability of your Kafka cluster.

11. Manual Replica Reassignment

Manual replica reassignment allows you to move replicas of a topic's partitions to different brokers in a controlled manner. This is useful for rebalancing load or migrating partitions.

Property	Description	Values
--topics-to-move-json-file	Path to a JSON file containing the topics and partitions to move.	`topics-to-move.json`
--reassignment-json-file	Path to a JSON file specifying the new replica assignments.	`reassignment.json`
--broker-list	Comma-separated list of brokers to which replicas should be reassigned.	`1,2,3`

Example: Reassigning Partitions

kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics-to-move.json --reassignment-json-file reassignment.json --execute

Example: JSON file for topics to move

{
  "topics": [
    {
      "topic": "my-topic"
    }
  ],
  "version": 1
}

Example: JSON file for reassignment

{
  "version": 1,
  "partitions": [
    {
      "topic": "my-topic",
      "partition": 0,
      "replicas": [1, 2, 3]
    }
  ]
}

Note: The reassignment process is asynchronous. Use kafka-reassign-partitions.sh --verify to check the status of ongoing reassignments.