Kafka：增加Topic的分区副本数

Posted 2022-02-20 ITKaven

测试环境参考下面这篇博客：

• Kafka：搭建Kafka集群

pom.xml：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.kaven</groupId>
    <artifactId>kafka</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.0.0</version>
        </dependency>
    </dependencies>
</project>

测试代码：

package com.kaven.kafka;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

public class Admin 

    // 基于Kafka服务地址与请求超时时间来创建AdminClient实例
    private static final AdminClient adminClient = Admin.getAdminClient("192.168.1.7:9092", "40000");

    public static void main(String[] args) throws InterruptedException, ExecutionException 
        Admin admin = new Admin();
        // 创建Topic，Topic名称为new-topic-user，分区副本数为1，复制因子为1
        admin.createTopic("new-topic-user", 1, (short) 1);
        // 获取指定Topic的描述
        admin.describeTopic("new-topic-user");
        // 增加指定Topic的分区副本数（从1增加到3）
        admin.addNumPartitions("new-topic-user", 3);
        // 获取指定Topic的描述
        admin.describeTopic("new-topic-user");
        Thread.sleep(10000);
    

    public static AdminClient getAdminClient(String address, String requestTimeoutMS) 
        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, address);
        properties.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMS);
        return AdminClient.create(properties);
    

    public void createTopic(String name, int numPartitions, short replicationFactor) throws InterruptedException 
        CountDownLatch latch = new CountDownLatch(1);
        CreateTopicsResult topics = adminClient.createTopics(
                Collections.singleton(new NewTopic(name, numPartitions, replicationFactor))
        );
        Map<String, KafkaFuture<Void>> values = topics.values();
        values.forEach((name__, future) -> 
            future.whenComplete((a, throwable) -> 
                if(throwable != null) 
                    System.out.println(throwable.getMessage());
                
                System.out.println(name__);
                latch.countDown();
            );
        );
        latch.await();
    

    public void describeTopic(String name) 
        Map<String, KafkaFuture<TopicDescription>> values =
                adminClient.describeTopics(Collections.singleton(name)).values();
        for (String name__ : values.keySet()) 
            values.get(name__).whenComplete((describe, throwable) -> 
                if(throwable != null) 
                    System.out.println(throwable.getMessage());
                
                System.out.println(name__);
                System.out.println(describe);
            );
        
    

    public void addNumPartitions(String name, int numPartitions) throws ExecutionException, InterruptedException 
        Map<String, NewPartitions> partitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(numPartitions);
        partitionsMap.put(name, newPartitions);
        adminClient.createPartitions(partitionsMap).all().get();
    

输出：

new-topic-user
new-topic-user
(name=new-topic-user, internal=false, partitions=(partition=0, leader=192.168.1.199:9092 (id: 2 rack: null), replicas=192.168.1.199:9092 (id: 2 rack: null), isr=192.168.1.199:9092 (id: 2 rack: null)), authorizedOperations=null)
new-topic-user
(name=new-topic-user, internal=false, partitions=(partition=0, leader=192.168.1.199:9092 (id: 2 rack: null), replicas=192.168.1.199:9092 (id: 2 rack: null), isr=192.168.1.199:9092 (id: 2 rack: null)),(partition=1, leader=192.168.1.7:9092 (id: 0 rack: null), replicas=192.168.1.7:9092 (id: 0 rack: null), isr=192.168.1.7:9092 (id: 0 rack: null)),(partition=2, leader=192.168.1.6:9092 (id: 1 rack: null), replicas=192.168.1.6:9092 (id: 1 rack: null), isr=192.168.1.6:9092 (id: 1 rack: null)), authorizedOperations=null)

很显然增加Topic的分区副本数成功了，那能不能减少Topic的分区副本数呢？

    public static void main(String[] args) throws InterruptedException, ExecutionException 
        Admin admin = new Admin();
        // 创建Topic，Topic名称为new-topic-user，分区副本数为1，复制因子为1
//        admin.createTopic("new-topic-user", 1, (short) 1);
        // 获取指定Topic的描述
        admin.describeTopic("new-topic-user");
        // 减少指定Topic的分区副本数（从3减少到2）
        admin.addNumPartitions("new-topic-user", 2);
        // 获取指定Topic的描述
        admin.describeTopic("new-topic-user");
        Thread.sleep(10000);
    

new-topic-user
(name=new-topic-user, internal=false, partitions=(partition=0, leader=192.168.1.199:9092 (id: 2 rack: null), replicas=192.168.1.199:9092 (id: 2 rack: null), isr=192.168.1.199:9092 (id: 2 rack: null)),(partition=1, leader=192.168.1.7:9092 (id: 0 rack: null), replicas=192.168.1.7:9092 (id: 0 rack: null), isr=192.168.1.7:9092 (id: 0 rack: null)),(partition=2, leader=192.168.1.6:9092 (id: 1 rack: null), replicas=192.168.1.6:9092 (id: 1 rack: null), isr=192.168.1.6:9092 (id: 1 rack: null)), authorizedOperations=null)
Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidPartitionsException: Topic currently has 3 partitions, which is higher than the requested 2.
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
	at com.kaven.kafka.Admin.addNumPartitions(Admin.java:74)
	at com.kaven.kafka.Admin.main(Admin.java:25)
Caused by: org.apache.kafka.common.errors.InvalidPartitionsException: Topic currently has 3 partitions, which is higher than the requested 2.

Topic当前有3个分区，高于请求的2个分区（Topic currently has 3 partitions, which is higher than the requested 2），很显然不能减少Topic的分区副本数。博客就到这里，如果博主有说错的地方或者大家有不同的见解，欢迎大家评论补充。