opensource4you · chaohengstudent · May 17, 2023 · May 17, 2023
diff --git a/app/src/main/java/org/astraea/app/backup/Backup.java b/app/src/main/java/org/astraea/app/backup/Backup.java
@@ -51,7 +51,7 @@ public void restoreDistribution(ClusterInfo clusterInfo, String bootstrapServers
                                                           Comparator.comparing(
                                                               replica -> !replica.isLeader()))
                                                       .map(replica -> replica.nodeInfo().id())
-                                                      .collect(Collectors.toUnmodifiableList()))))))
+                                                      .toList())))))
                       .configs(topic.config().raw())
                       .run()
                       .toCompletableFuture()

diff --git a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
@@ -73,9 +73,7 @@ static List<ConsumerThread> create(
           consumerSupplier) {
     if (consumers == 0) return List.of();
     var closeLatches =
-        IntStream.range(0, consumers)
-            .mapToObj(ignored -> new CountDownLatch(1))
-            .collect(Collectors.toUnmodifiableList());
+        IntStream.range(0, consumers).mapToObj(ignored -> new CountDownLatch(1)).toList();
     var executors = Executors.newFixedThreadPool(consumers);
     // monitor
     CompletableFuture.runAsync(

diff --git a/app/src/main/java/org/astraea/app/performance/DataGenerator.java b/app/src/main/java/org/astraea/app/performance/DataGenerator.java
@@ -26,7 +26,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 import org.astraea.common.Configuration;
 import org.astraea.common.DataUnit;
@@ -49,15 +48,13 @@ static DataGenerator of(
         RecordGenerator.builder()
             .batchSize(argument.transactionSize)
             .keyTableSeed(argument.recordKeyTableSeed)
-            .keyRange(
-                LongStream.rangeClosed(0, 10000).boxed().collect(Collectors.toUnmodifiableList()))
+            .keyRange(LongStream.rangeClosed(0, 10000).boxed().toList())
             .keyDistribution(argument.keyDistributionType.create(10000, keyDistConfig))
             .keySizeDistribution(
                 argument.keySizeDistributionType.create(
                     (int) argument.keySize.bytes(), keySizeDistConfig))
             .valueTableSeed(argument.recordValueTableSeed)
-            .valueRange(
-                LongStream.rangeClosed(0, 10000).boxed().collect(Collectors.toUnmodifiableList()))
+            .valueRange(LongStream.rangeClosed(0, 10000).boxed().toList())
             .valueDistribution(argument.valueDistributionType.create(10000, valueDistConfig))
             .valueSizeDistribution(
                 argument.valueDistributionType.create(

diff --git a/app/src/main/java/org/astraea/app/performance/MonkeyThread.java b/app/src/main/java/org/astraea/app/performance/MonkeyThread.java
@@ -22,7 +22,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
 import org.astraea.common.Utils;
 import org.astraea.common.consumer.Consumer;
 import org.astraea.common.consumer.ConsumerConfigs;
@@ -52,7 +51,7 @@ static List<MonkeyThread> play(List<ConsumerThread> consumerThreads, Performance
                   return unsubscribeMonkey(consumerThreads, entry.getValue());
               }
             })
-        .collect(Collectors.toUnmodifiableList());
+        .toList();
   }
 
   private static MonkeyThread killMonkey(List<ConsumerThread> consumerThreads, Duration frequency) {

diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -76,7 +76,7 @@ public static List<String> execute(final Argument param) {
     var blockingQueues =
         IntStream.range(0, param.producers)
             .mapToObj(i -> new ArrayBlockingQueue<List<Record<byte[], byte[]>>>(3000))
-            .collect(Collectors.toUnmodifiableList());
+            .toList();
     // ensure topics are existent
     System.out.println("checking topics: " + String.join(",", param.topics));
     param.checkTopics();
@@ -123,10 +123,7 @@ public static List<String> execute(final Argument param) {
             var current = Report.recordsConsumedTotal();
 
             if (blockingQueues.stream().allMatch(Collection::isEmpty)) {
-              var unfinishedProducers =
-                  producerThreads.stream()
-                      .filter(p -> !p.closed())
-                      .collect(Collectors.toUnmodifiableList());
+              var unfinishedProducers = producerThreads.stream().filter(p -> !p.closed()).toList();
               unfinishedProducers.forEach(AbstractThread::close);
             }
 
@@ -388,7 +385,7 @@ else if (specifiedByBroker) {
                   .filter(replica -> specifyBrokers.contains(replica.nodeInfo().id()))
                   .map(replica -> TopicPartition.of(replica.topic(), replica.partition()))
                   .distinct()
-                  .collect(Collectors.toUnmodifiableList());
+                  .toList();
           if (selections.isEmpty())
             throw new IllegalArgumentException(
                 "No partition match the specify.brokers requirement");
@@ -426,8 +423,7 @@ else if (specifiedByBroker) {
                 "The following topic/partitions are nonexistent in the cluster: " + notExist);
         }
 
-        final var selection =
-            specifyPartitions.stream().distinct().collect(Collectors.toUnmodifiableList());
+        final var selection = specifyPartitions.stream().distinct().toList();
         return () -> selection.get(ThreadLocalRandom.current().nextInt(selection.size()));
       } else if (throttle) {
         // TODO: The functions of throttle and select partitioner should not conflict with each
@@ -444,15 +440,12 @@ else if (specifiedByBroker) {
                   .replicaStream()
                   .map(Replica::topicPartition)
                   .distinct()
-                  .collect(Collectors.toUnmodifiableList());
+                  .toList();
           return () -> selection.get(ThreadLocalRandom.current().nextInt(selection.size()));
         }
       } else {
         final var selection =
-            topics.stream()
-                .map(topic -> TopicPartition.of(topic, -1))
-                .distinct()
-                .collect(Collectors.toUnmodifiableList());
+            topics.stream().map(topic -> TopicPartition.of(topic, -1)).distinct().toList();
         return () -> selection.get(ThreadLocalRandom.current().nextInt(selection.size()));
       }
     }

diff --git a/app/src/main/java/org/astraea/app/performance/ProducerThread.java b/app/src/main/java/org/astraea/app/performance/ProducerThread.java
@@ -51,9 +51,7 @@ static List<ProducerThread> create(
     var producers = queues.size();
     if (producers <= 0) return List.of();
     var closeLatches =
-        IntStream.range(0, producers)
-            .mapToObj(ignored -> new CountDownLatch(1))
-            .collect(Collectors.toUnmodifiableList());
+        IntStream.range(0, producers).mapToObj(ignored -> new CountDownLatch(1)).toList();
     var executors = Executors.newFixedThreadPool(producers);
     // monitor
     CompletableFuture.runAsync(

diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -179,7 +179,7 @@ private PlanExecutionProgress progress(String taskId) {
                         tp ->
                             Change.from(
                                 contextCluster.replicas(tp), solution.proposal().replicas(tp)))
-                    .collect(Collectors.toUnmodifiableList());
+                    .toList();
     var report =
         (Supplier<PlanReport>)
             () ->

diff --git a/app/src/main/java/org/astraea/app/web/BeanHandler.java b/app/src/main/java/org/astraea/app/web/BeanHandler.java
@@ -19,7 +19,6 @@
 import java.util.List;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.metrics.BeanObject;
 import org.astraea.common.metrics.BeanQuery;
@@ -48,12 +47,10 @@ public CompletionStage<Response> get(Channel channel) {
                               try (var client = JndiClient.of(b.host(), jmxPorts.apply(b.id()))) {
                                 return new NodeBean(
                                     b.host(),
-                                    client.beans(builder.build()).stream()
-                                        .map(Bean::new)
-                                        .collect(Collectors.toUnmodifiableList()));
+                                    client.beans(builder.build()).stream().map(Bean::new).toList());
                               }
                             })
-                        .collect(Collectors.toUnmodifiableList())));
+                        .toList()));
   }
 
   static class Property implements Response {
@@ -86,11 +83,11 @@ static class Bean implements Response {
       this.properties =
           obj.properties().entrySet().stream()
               .map(e -> new Property(e.getKey(), e.getValue()))
-              .collect(Collectors.toUnmodifiableList());
+              .toList();
       this.attributes =
           obj.attributes().entrySet().stream()
               .map(e -> new Attribute(e.getKey(), e.getValue().toString()))
-              .collect(Collectors.toUnmodifiableList());
+              .toList();
     }
   }
 

diff --git a/app/src/main/java/org/astraea/app/web/BrokerHandler.java b/app/src/main/java/org/astraea/app/web/BrokerHandler.java
@@ -64,7 +64,7 @@ public CompletionStage<Response> get(Channel channel) {
                             brokers.stream()
                                 .filter(b -> ids.contains(b.id()))
                                 .map(Broker::new)
-                                .collect(Collectors.toList())))
+                                .toList()))
         .thenApply(
             brokers -> {
               if (brokers.isEmpty()) throw new NoSuchElementException("no brokers are found");
@@ -96,7 +96,7 @@ static class Broker implements Response {
               .entrySet()
               .stream()
               .map(e -> new Topic(e.getKey(), e.getValue().size()))
-              .collect(Collectors.toUnmodifiableList());
+              .toList();
       this.configs = broker.config().raw();
     }
   }

diff --git a/app/src/main/java/org/astraea/app/web/Channel.java b/app/src/main/java/org/astraea/app/web/Channel.java
@@ -159,7 +159,7 @@ static Channel of(HttpExchange exchange) {
               Arrays.stream(uri.getPath().split("/"))
                   .map(String::trim)
                   .filter(s -> !s.isEmpty())
-                  .collect(Collectors.toUnmodifiableList());
+                  .toList();
           // form: /resource/target
           if (allPaths.size() == 1) return Optional.empty();
           else if (allPaths.size() == 2) return Optional.of(allPaths.get(1));

diff --git a/app/src/main/java/org/astraea/app/web/GroupHandler.java b/app/src/main/java/org/astraea/app/web/GroupHandler.java
@@ -144,10 +144,9 @@ public CompletionStage<Response> get(Channel channel) {
                                                                                 empty())
                                                             .filter(Optional::isPresent)
                                                             .map(Optional::get)
-                                                            .collect(
-                                                                Collectors.toUnmodifiableList())))
-                                            .collect(Collectors.toUnmodifiableList())))
-                            .collect(Collectors.toUnmodifiableList())))
+                                                            .toList()))
+                                            .toList()))
+                            .toList()))
         .thenApply(
             groups -> {
               if (channel.target().isPresent() && groups.size() == 1) return groups.get(0);

diff --git a/app/src/main/java/org/astraea/app/web/ProducerHandler.java b/app/src/main/java/org/astraea/app/web/ProducerHandler.java
@@ -60,7 +60,7 @@ public CompletionStage<Partitions> get(Channel channel) {
                         .entrySet()
                         .stream()
                         .map(e -> new Partition(e.getKey(), e.getValue()))
-                        .collect(Collectors.toUnmodifiableList())));
+                        .toList()));
   }
 
   static class ProducerState implements Response {
@@ -88,8 +88,7 @@ static class Partition implements Response {
         Collection<org.astraea.common.admin.ProducerState> states) {
       this.topic = tp.topic();
       this.partition = tp.partition();
-      this.states =
-          states.stream().map(ProducerState::new).collect(Collectors.toUnmodifiableList());
+      this.states = states.stream().map(ProducerState::new).toList();
     }
   }
 

diff --git a/app/src/main/java/org/astraea/app/web/QuotaHandler.java b/app/src/main/java/org/astraea/app/web/QuotaHandler.java
@@ -167,7 +167,7 @@ static class Quotas implements Response {
     final List<Quota> quotas;
 
     Quotas(Collection<org.astraea.common.admin.Quota> quotas) {
-      this.quotas = quotas.stream().map(Quota::new).collect(Collectors.toUnmodifiableList());
+      this.quotas = quotas.stream().map(Quota::new).toList();
     }
   }
 

diff --git a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
@@ -90,9 +90,7 @@ public CompletionStage<Response> post(Channel channel) {
                               if (excludedBroker.isEmpty())
                                 return CompletableFuture.completedFuture(Response.BAD_REQUEST);
                               var availableBrokers =
-                                  brokers.stream()
-                                      .filter(b -> b.id() != exclude)
-                                      .collect(Collectors.toList());
+                                  brokers.stream().filter(b -> b.id() != exclude).toList();
                               var partitions =
                                   excludedBroker.get().topicPartitions().stream()
                                       .filter(
@@ -131,7 +129,7 @@ public CompletionStage<Response> post(Channel channel) {
             Stream.of(process2Folders, process2Nodes, processExclude)
                 .flatMap(Function.identity())
                 .map(CompletionStage::toCompletableFuture)
-                .collect(Collectors.toUnmodifiableList()))
+                .toList())
         .thenApply(
             rs -> {
               if (!rs.isEmpty() && rs.stream().allMatch(r -> r == Response.ACCEPT))
@@ -161,7 +159,7 @@ public CompletionStage<Reassignments> get(Channel channel) {
                           r ->
                               new AddingReplica(
                                   r, leaderSizes.getOrDefault(r.topicPartition(), 0L)))
-                      .collect(Collectors.toUnmodifiableList()));
+                      .toList());
             });
   }
 

diff --git a/app/src/main/java/org/astraea/app/web/TopicHandler.java b/app/src/main/java/org/astraea/app/web/TopicHandler.java
@@ -116,7 +116,7 @@ private CompletionStage<Topics> get(
                                           .filter(replica -> replica.topic().equals(p.topic()))
                                           .filter(replica -> replica.partition() == p.partition())
                                           .map(Replica::new)
-                                          .collect(Collectors.toUnmodifiableList())),
+                                          .toList()),
                               Collectors.toList())));
           // topic name -> group ids
           var gs =
@@ -149,7 +149,7 @@ private CompletionStage<Topics> get(
                               gs.getOrDefault(topic.name(), Set.of()),
                               ps.get(topic.name()),
                               topic.config().raw()))
-                  .collect(Collectors.toUnmodifiableList()));
+                  .toList());
         });
   }
 
@@ -197,7 +197,7 @@ public CompletionStage<Topics> post(Channel channel) {
                 new Topics(
                     topicNames.stream()
                         .map(t -> new TopicInfo(t, Set.of(), List.of(), Map.of()))
-                        .collect(Collectors.toUnmodifiableList())));
+                        .toList()));
   }
 
   @Override

diff --git a/app/src/main/java/org/astraea/app/web/TransactionHandler.java b/app/src/main/java/org/astraea/app/web/TransactionHandler.java
@@ -49,9 +49,7 @@ public CompletionStage<Response> get(Channel channel) {
         .thenCompose(admin::transactions)
         .thenApply(
             transactions ->
-                transactions.stream()
-                    .map(t -> new Transaction(t.transactionId(), t))
-                    .collect(Collectors.toUnmodifiableList()))
+                transactions.stream().map(t -> new Transaction(t.transactionId(), t)).toList())
         .thenApply(
             transactions -> {
               if (channel.target().isPresent() && transactions.size() == 1)