測試環境參考下面這篇部落格:
- 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)
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.