garyparrot
commented
1 year ago 這邊敘述建立情境的方法
建立叢集
下載下面這個檔案，解壓縮後裡面有一個序列化後的 ClusterInfo exp2-cluster-info-before-greedy.bin.zip
用 Astraea 專案內的 library 可以反序列化裡面的資訊，之後要將裡面描述的叢集套用到一個 6 節點的 Kafka Cluster，其中每個節點 ID 必須要是 0,1,2,3,4,5。
下面是一個套用的範例程式碼
```java
@Test
void restoreClusterInfo() throws IOException {
var file = "/path/to/exp2-cluster-info-before-greedy.bin";
try (
var admin = Admin.of(bootstrap);
var stream = Files.newInputStream(Path.of(file))) {
var cluster = ByteUtils.readClusterInfo(stream.readAllBytes());
System.out.println("Delete Topics");
admin.topicNames(false)
.thenApply(x -> x.stream()
.filter(xx -> !xx.startsWith("__"))
.collect(Collectors.toSet()))
.thenApply(x -> {
System.out.println("Delete: " + x);
return x;
})
.thenCompose(admin::deleteTopics)
.toCompletableFuture()
.join();
Utils.sleep(Duration.ofSeconds(10));
System.out.println("Recreate Topics");
cluster.replicas().stream()
.collect(Collectors.groupingBy(Replica::topic, Collectors.mapping(Replica::topicPartition, Collectors.counting())))
.entrySet()
.stream()
.map(x -> admin.creator()
.topic(x.getKey())
.numberOfPartitions(x.getValue().intValue())
.numberOfReplicas((short)1)
.run()
.toCompletableFuture())
.toList()
.forEach(CompletableFuture::join);
System.out.println("Relocate Replicas");
admin.moveToBrokers(cluster.topicPartitions()
.stream()
.collect(Collectors.toUnmodifiableMap(
tp -> tp,
tp -> cluster.replicas(tp).stream()
.sorted(Comparator.comparing(x -> !x.isPreferredLeader()))
.map(Replica::brokerId)
.collect(Collectors.toList())
)))
.toCompletableFuture()
.join();
Utils.sleep(Duration.ofSeconds(5));
System.out.println("Relocate to Folders");
admin.moveToFolders(cluster.replicas()
.stream()
.collect(Collectors.toUnmodifiableMap(Replica::topicPartitionReplica, Replica::path)))
.toCompletableFuture()
.join();
Utils.sleep(Duration.ofSeconds(5));
System.out.println("Leader election");
admin.preferredLeaderElection(cluster.topicPartitions());
Utils.sleep(Duration.ofSeconds(5));
admin.topicNames(true)
.thenCompose(admin::clusterInfo)
.toCompletableFuture()
.join()
.replicas()
.stream()
.filter(x -> x.isFuture() || x.isAdding() || x.isRemoving())
.forEach(System.out::println);
}
}
```
建立 Producer & Consumer
exp2-ansible.txt 上述文字檔案裡面描述 Producer & Consumer 的設定，總共會需要 7 台獨立的設備，每一台執行一個 Astraea Performance Tool
每個欄位會對應到 Performance Tool 的參數
- "./bin/app"
- "performance"
- "--bootstrap.servers {{ bootstrap_servers }}"
- "--run.until 3h"
- "--topics {{ topics }}"
- "{{ throttle if throttle_enable else '' }}"
- "{{ throughput if throughput_enable else '' }}"
- "--record.key.table.seed {{ key_table_seed }}"
- "--key.distribution {{ key_distribution }}"
- "--key.distribution.config '{{ key_distribution_config | default('k=v') }}'"
- "--read.idle 15s"
- "--producers {{ 0 if (no_producer=='true') else 4 }}"
- "--consumers {{ 0 if (no_consumer=='true') else 12 }}"
- "--configs receive.buffer.bytes=1002400,buffer.memory=500000000,linger.ms=1,batch.size=1000000,fetch.max.bytes=50000000,fetch.min.bytes=20000000,max.partition.fetch.bytes=20000000,fetch.max.wait.ms=5,max.poll.records=9999999,check.crcs=false,acks=0,metrics.sample.window.ms=5000"流量的趨勢
完成上述步驟後叢集會有這樣的流量趨勢：
節點的輸入
節點的輸出
如果趨勢不符合的話代表有地方重現失敗
brandboat
比較 Balancer 的優化效果