Kafka — Rebalance

In this example, we’ll make the use of kafka rebalance.

Create the topic with just 1 partition.

kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic t-rebalance
Created topic t-rebalance.

Start the Producer and consumer

RebalanceProducer.java

import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class RebalanceProducer {

   @Autowired
   private KafkaTemplate<String, String> kafkaTemplate;

   private AtomicInteger counter = new AtomicInteger();

   @Scheduled(fixedRate = 1000)
   public void sendMessage() {
      kafkaTemplate.send("t-rebalance", "Counter " + counter.incrementAndGet());
   }
}

MainApp.java

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class KafkaProducerApplication implements CommandLineRunner {

   public static void main(String[] args) {
      SpringApplication.run(KafkaProducerApplication.class, args);
   }

   @Override
   public void run(String... args) throws Exception {

   }
}

RebalanceConsumer.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class RebalanceConsumer {

   private static final Logger LOG = LoggerFactory.getLogger(RebalanceConsumer.class);

   @KafkaListener(topics = "t-rebalance", concurrency = "3")
   public void consume(ConsumerRecord<String, String> consumerRecord) {
      LOG.info("Partition : {}, Offset : {}, Message : {}", consumerRecord.partition(), consumerRecord.offset(),
            consumerRecord.value());
   }
}

At this point only one partition where data will be saved! now you alter the partition using below command.

kafka-topics --bootstrap-server localhost:9092 --alter --topic t-rebalance --partitions 2

Almost after 3 minutes you’ll see that data will be loaded into partition-2. Even you can alter one more time and create one more partitions.