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-rebalanceCreated 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.

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store