[CONNECTOR] rewrite sink connector by java 17 toList (#1754)

opensource4you · May 17, 2023 · 94aa36a · 94aa36a
1 parent 4d0c726
commit 94aa36a
Show file tree

Hide file tree

Showing 4 changed files with 4 additions and 10 deletions.
diff --git a/connector/src/main/java/org/astraea/connector/SinkConnector.java b/connector/src/main/java/org/astraea/connector/SinkConnector.java
@@ -18,7 +18,6 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.kafka.common.config.ConfigDef;
 import org.astraea.common.Configuration;
 import org.astraea.common.VersionUtils;
@@ -53,9 +52,7 @@ public final Class<? extends org.apache.kafka.connect.sink.SinkTask> taskClass()
 
   @Override
   public List<Map<String, String>> taskConfigs(int maxTasks) {
-    return takeConfiguration(maxTasks).stream()
-        .map(Configuration::raw)
-        .collect(Collectors.toList());
+    return takeConfiguration(maxTasks).stream().map(Configuration::raw).toList();
   }
 
   @Override

diff --git a/connector/src/main/java/org/astraea/connector/SinkTask.java b/connector/src/main/java/org/astraea/connector/SinkTask.java
@@ -21,7 +21,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.header.Headers;
@@ -57,7 +56,7 @@ public final void start(Map<String, String> props) {
   @Override
   public final void put(Collection<SinkRecord> records) {
     if (records != null && !records.isEmpty())
-      put(records.stream().map(SinkTask::toRecord).collect(Collectors.toList()));
+      put(records.stream().map(SinkTask::toRecord).toList());
   }
 
   private static byte[] toBytes(Schema schema, Object value) {

diff --git a/connector/src/main/java/org/astraea/connector/backup/Exporter.java b/connector/src/main/java/org/astraea/connector/backup/Exporter.java
@@ -26,7 +26,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.LongAdder;
-import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.common.Configuration;
@@ -128,7 +127,7 @@ protected Class<? extends SinkTask> task() {
 
   @Override
   protected List<Configuration> takeConfiguration(int maxTasks) {
-    return IntStream.range(0, maxTasks).mapToObj(ignored -> configs).collect(Collectors.toList());
+    return IntStream.range(0, maxTasks).mapToObj(ignored -> configs).toList();
   }
 
   @Override

diff --git a/connector/src/main/java/org/astraea/connector/perf/PerfSink.java b/connector/src/main/java/org/astraea/connector/perf/PerfSink.java
@@ -18,7 +18,6 @@
 
 import java.time.Duration;
 import java.util.List;
-import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.common.Configuration;
 import org.astraea.common.Utils;
@@ -50,7 +49,7 @@ protected Class<? extends SinkTask> task() {
 
   @Override
   protected List<Configuration> takeConfiguration(int maxTasks) {
-    return IntStream.range(0, maxTasks).mapToObj(i -> config).collect(Collectors.toList());
+    return IntStream.range(0, maxTasks).mapToObj(i -> config).toList();
   }
 
   @Override