Skip to content

Commit 00a6fc7

Browse files
qoo332001qoo332001
committed
fix bugs
2 parents 85ce98d + ced840f commit 00a6fc7

84 files changed

Lines changed: 848 additions & 1011 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎app/src/main/java/org/astraea/app/backup/Backup.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public void restoreDistribution(ClusterInfo clusterInfo, String bootstrapServers
5050
.sorted(
5151
Comparator.comparing(
5252
replica -> !replica.isLeader()))
53-
.map(replica -> replica.nodeInfo().id())
53+
.map(replica -> replica.broker().id())
5454
.toList())))))
5555
.configs(topic.config().raw())
5656
.run()

‎app/src/main/java/org/astraea/app/performance/Performance.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -382,7 +382,7 @@ else if (specifiedByBroker) {
382382
.join()
383383
.replicaStream()
384384
.filter(Replica::isLeader)
385-
.filter(replica -> specifyBrokers.contains(replica.nodeInfo().id()))
385+
.filter(replica -> specifyBrokers.contains(replica.broker().id()))
386386
.map(replica -> TopicPartition.of(replica.topic(), replica.partition()))
387387
.distinct()
388388
.toList();

‎app/src/main/java/org/astraea/app/publisher/MetricPublisher.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import org.astraea.app.argument.StringMapField;
2626
import org.astraea.common.Utils;
2727
import org.astraea.common.admin.Admin;
28-
import org.astraea.common.admin.NodeInfo;
28+
import org.astraea.common.admin.Broker;
2929
import org.astraea.common.metrics.JndiClient;
3030
import org.astraea.common.metrics.collector.MetricFetcher;
3131

@@ -50,17 +50,17 @@ static void execute(Arguments arguments) {
5050
.clientSupplier(
5151
() ->
5252
admin
53-
.nodeInfos()
53+
.brokers()
5454
.thenApply(
55-
nodes ->
56-
nodes.stream()
55+
brokers ->
56+
brokers.stream()
5757
.collect(
5858
Collectors.toUnmodifiableMap(
59-
NodeInfo::id,
60-
node ->
59+
Broker::id,
60+
broker ->
6161
JndiClient.of(
62-
node.host(),
63-
arguments.idToJmxPort().apply(node.id()))))))
62+
broker.host(),
63+
arguments.idToJmxPort().apply(broker.id()))))))
6464
.fetchBeanDelay(arguments.period)
6565
.fetchMetadataDelay(Duration.ofMinutes(5))
6666
.threads(3)

‎app/src/main/java/org/astraea/app/web/BalancerHandler.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -256,7 +256,7 @@ static class Placement {
256256
final Optional<Long> size;
257257

258258
Placement(Replica replica, Optional<Long> size) {
259-
this.brokerId = replica.nodeInfo().id();
259+
this.brokerId = replica.broker().id();
260260
this.directory = replica.path();
261261
this.size = size;
262262
}

‎app/src/main/java/org/astraea/app/web/BrokerHandler.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
import java.util.concurrent.CompletionStage;
2626
import java.util.stream.Collectors;
2727
import org.astraea.common.admin.Admin;
28-
import org.astraea.common.admin.NodeInfo;
28+
import org.astraea.common.admin.Broker;
2929
import org.astraea.common.admin.TopicPartition;
3030

3131
class BrokerHandler implements Handler {
@@ -44,8 +44,12 @@ CompletionStage<Set<Integer>> brokers(Optional<String> target) {
4444
.orElseGet(
4545
() ->
4646
admin
47-
.nodeInfos()
48-
.thenApply(ns -> ns.stream().map(NodeInfo::id).collect(Collectors.toSet())));
47+
.brokers()
48+
.thenApply(
49+
ns ->
50+
ns.stream()
51+
.map(org.astraea.common.admin.Broker::id)
52+
.collect(Collectors.toSet())));
4953
} catch (NumberFormatException e) {
5054
return CompletableFuture.failedFuture(
5155
new NoSuchElementException("the broker id must be number"));

‎app/src/main/java/org/astraea/app/web/ReassignmentHandler.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,8 @@
3131
import java.util.stream.Stream;
3232
import org.astraea.common.FutureUtils;
3333
import org.astraea.common.admin.Admin;
34+
import org.astraea.common.admin.Broker;
3435
import org.astraea.common.admin.ClusterInfo;
35-
import org.astraea.common.admin.NodeInfo;
3636
import org.astraea.common.admin.Replica;
3737
import org.astraea.common.admin.TopicPartition;
3838
import org.astraea.common.admin.TopicPartitionReplica;
@@ -110,7 +110,7 @@ public CompletionStage<Response> post(Channel channel) {
110110
availableBrokers.stream()
111111
.filter(
112112
b -> b.topicPartitions().contains(tp))
113-
.map(NodeInfo::id)
113+
.map(Broker::id)
114114
.toList();
115115
if (!ids.isEmpty()) return ids;
116116
return List.of(
@@ -211,7 +211,7 @@ static class AddingReplica implements Response {
211211
AddingReplica(Replica addingReplica, long leaderSize) {
212212
this.topicName = addingReplica.topic();
213213
this.partition = addingReplica.partition();
214-
this.broker = addingReplica.nodeInfo().id();
214+
this.broker = addingReplica.broker().id();
215215
this.dataFolder = addingReplica.path();
216216
this.size = addingReplica.size();
217217
this.leaderSize = leaderSize;

‎app/src/main/java/org/astraea/app/web/SkewedPartitionScenario.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
import org.apache.commons.math3.distribution.IntegerDistribution;
3131
import org.apache.commons.math3.util.Pair;
3232
import org.astraea.common.admin.Admin;
33-
import org.astraea.common.admin.NodeInfo;
33+
import org.astraea.common.admin.Broker;
3434
import org.astraea.common.admin.TopicPartition;
3535
import org.astraea.common.admin.TopicPartitionReplica;
3636

@@ -62,7 +62,7 @@ public CompletionStage<Result> apply(Admin admin) {
6262
admin.waitPartitionLeaderSynced(
6363
Map.of(topicName, partitions), Duration.ofSeconds(4)))
6464
.thenCompose(ignored -> admin.brokers())
65-
.thenApply(brokers -> brokers.stream().map(NodeInfo::id).sorted().toList())
65+
.thenApply(brokers -> brokers.stream().map(Broker::id).sorted().toList())
6666
.thenCompose(
6767
brokerIds -> {
6868
var distribution =

‎app/src/main/java/org/astraea/app/web/ThrottleHandler.java‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@
3030
import org.astraea.common.EnumInfo;
3131
import org.astraea.common.FutureUtils;
3232
import org.astraea.common.admin.Admin;
33+
import org.astraea.common.admin.Broker;
3334
import org.astraea.common.admin.BrokerConfigs;
34-
import org.astraea.common.admin.NodeInfo;
3535
import org.astraea.common.admin.TopicConfigs;
3636
import org.astraea.common.admin.TopicPartitionReplica;
3737
import org.astraea.common.json.TypeRef;
@@ -100,9 +100,8 @@ public CompletionStage<Response> post(Channel channel) {
100100

101101
var topicToAppends =
102102
admin
103-
.nodeInfos()
104-
.thenApply(
105-
nodeInfos -> nodeInfos.stream().map(NodeInfo::id).collect(Collectors.toSet()))
103+
.brokers()
104+
.thenApply(brokers -> brokers.stream().map(Broker::id).collect(Collectors.toSet()))
106105
.thenCompose(admin::topicPartitionReplicas)
107106
.thenApply(
108107
replicas ->
@@ -197,9 +196,8 @@ public CompletionStage<Response> delete(Channel channel) {
197196

198197
var topicToSubtracts =
199198
admin
200-
.nodeInfos()
201-
.thenApply(
202-
nodeInfos -> nodeInfos.stream().map(NodeInfo::id).collect(Collectors.toSet()))
199+
.brokers()
200+
.thenApply(brokers -> brokers.stream().map(Broker::id).collect(Collectors.toSet()))
203201
.thenCompose(admin::topicPartitionReplicas)
204202
.thenApply(
205203
replicas -> {
@@ -251,11 +249,11 @@ public CompletionStage<Response> delete(Channel channel) {
251249

252250
var brokerToUnset =
253251
admin
254-
.nodeInfos()
252+
.brokers()
255253
.thenApply(
256254
ns ->
257255
ns.stream()
258-
.map(NodeInfo::id)
256+
.map(Broker::id)
259257
.filter(
260258
id ->
261259
!channel.queries().containsKey("broker")

‎app/src/main/java/org/astraea/app/web/TopicHandler.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -316,7 +316,7 @@ private Replica() {
316316

317317
Replica(org.astraea.common.admin.Replica replica) {
318318
this(
319-
replica.nodeInfo().id(),
319+
replica.broker().id(),
320320
replica.lag(),
321321
replica.size(),
322322
replica.isLeader(),

‎app/src/main/java/org/astraea/app/web/WebService.java‎

Lines changed: 49 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -24,24 +24,30 @@
2424
import java.util.Collection;
2525
import java.util.List;
2626
import java.util.Map;
27-
import java.util.concurrent.CompletionStage;
27+
import java.util.concurrent.CompletableFuture;
2828
import java.util.concurrent.ConcurrentLinkedQueue;
2929
import java.util.concurrent.TimeUnit;
30+
import java.util.function.BiConsumer;
3031
import java.util.function.Function;
3132
import java.util.function.Supplier;
3233
import java.util.stream.Collectors;
3334
import org.astraea.app.argument.DurationField;
3435
import org.astraea.app.argument.IntegerMapField;
3536
import org.astraea.app.argument.NonNegativeIntegerField;
37+
import org.astraea.common.Configuration;
3638
import org.astraea.common.Utils;
3739
import org.astraea.common.admin.Admin;
38-
import org.astraea.common.admin.NodeInfo;
40+
import org.astraea.common.admin.Broker;
3941
import org.astraea.common.metrics.JndiClient;
4042
import org.astraea.common.metrics.MBeanClient;
4143
import org.astraea.common.metrics.collector.MetricSensor;
4244
import org.astraea.common.metrics.collector.MetricStore;
4345

4446
public class WebService implements AutoCloseable {
47+
public static final String METRIC_STORE_KEY = "metric.store";
48+
public static final String METRIC_STORE_LOCAL = "local";
49+
public static final String METRIC_STORE_TOPIC = "topic";
50+
public static final String BOOTSTRAP_SERVERS_KEY = "bootstrap.servers";
4551

4652
private final HttpServer server;
4753
private final Admin admin;
@@ -51,32 +57,48 @@ public WebService(
5157
Admin admin,
5258
int port,
5359
Function<Integer, Integer> brokerIdToJmxPort,
54-
Duration beanExpiration) {
60+
Duration beanExpiration,
61+
Configuration config) {
5562
this.admin = admin;
56-
Supplier<CompletionStage<Map<Integer, MBeanClient>>> clientSupplier =
63+
Supplier<Map<MetricSensor, BiConsumer<Integer, Exception>>> sensorsSupplier =
5764
() ->
58-
admin
59-
.brokers()
60-
.thenApply(
61-
brokers ->
62-
brokers.stream()
63-
.collect(
64-
Collectors.toUnmodifiableMap(
65-
NodeInfo::id,
66-
b ->
67-
JndiClient.of(b.host(), brokerIdToJmxPort.apply(b.id())))));
65+
sensors.metricSensors().stream()
66+
.distinct()
67+
.collect(
68+
Collectors.toUnmodifiableMap(Function.identity(), ignored -> (id, ee) -> {}));
69+
70+
List<MetricStore.Receiver> receivers =
71+
switch (config.string(METRIC_STORE_KEY).orElse(METRIC_STORE_LOCAL)) {
72+
case METRIC_STORE_LOCAL -> {
73+
Function<List<Broker>, Map<Integer, MBeanClient>> asBeanClientMap =
74+
brokers ->
75+
brokers.stream()
76+
.collect(
77+
Collectors.toUnmodifiableMap(
78+
Broker::id,
79+
b -> JndiClient.of(b.host(), brokerIdToJmxPort.apply(b.id()))));
80+
yield List.of(
81+
MetricStore.Receiver.local(() -> admin.brokers().thenApply(asBeanClientMap)));
82+
}
83+
case METRIC_STORE_TOPIC -> List.of(
84+
MetricStore.Receiver.topic(config.requireString(BOOTSTRAP_SERVERS_KEY)),
85+
MetricStore.Receiver.local(
86+
() -> CompletableFuture.completedStage(Map.of(-1, JndiClient.local()))));
87+
default -> throw new IllegalArgumentException(
88+
"unknown metric store type: "
89+
+ config.string(METRIC_STORE_KEY)
90+
+ ". use "
91+
+ METRIC_STORE_LOCAL
92+
+ " or "
93+
+ METRIC_STORE_TOPIC);
94+
};
6895
var metricStore =
6996
MetricStore.builder()
7097
.beanExpiration(beanExpiration)
71-
.receivers(List.of(MetricStore.Receiver.local(clientSupplier)))
72-
.sensorsSupplier(
73-
() ->
74-
sensors.metricSensors().stream()
75-
.distinct()
76-
.collect(
77-
Collectors.toUnmodifiableMap(
78-
Function.identity(), ignored -> (id, ee) -> {})))
98+
.receivers(receivers)
99+
.sensorsSupplier(sensorsSupplier)
79100
.build();
101+
80102
server = Utils.packException(() -> HttpServer.create(new InetSocketAddress(port), 0));
81103
server.createContext("/topics", to(new TopicHandler(admin)));
82104
server.createContext("/groups", to(new GroupHandler(admin)));
@@ -109,7 +131,11 @@ public static void main(String[] args) throws Exception {
109131
throw new IllegalArgumentException("you must define either --jmx.port or --jmx.ports");
110132
try (var service =
111133
new WebService(
112-
Admin.of(arg.configs()), arg.port, arg::jmxPortMapping, arg.beanExpiration)) {
134+
Admin.of(arg.configs()),
135+
arg.port,
136+
arg::jmxPortMapping,
137+
arg.beanExpiration,
138+
new Configuration(arg.configs()))) {
113139
if (arg.ttl == null) {
114140
System.out.println("enter ctrl + c to terminate web service");
115141
TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);

0 commit comments

Comments
 (0)