[CONNECTOR] add perf sink

connector/src/main/java/org/astraea/connector/perf/PerfSink.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.connector.perf;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.common.Configuration;
+import org.astraea.common.Utils;
+import org.astraea.connector.Definition;
+import org.astraea.connector.SinkConnector;
+import org.astraea.connector.SinkTask;
+
+public class PerfSink extends SinkConnector {
+
+  static Definition FREQUENCY_DEF =
+      Definition.builder()
+          .name("frequency.in.seconds")
+          .type(Definition.Type.STRING)
+          .defaultValue("300ms")
+          .validator((name, value) -> Utils.toDuration(value.toString()))
+          .build();
+
+  private Configuration config;
+
+  @Override
+  protected void init(Configuration configuration) {
+    this.config = configuration;
+  }
+
+  @Override
+  protected Class<? extends SinkTask> task() {
+    return PerfSinkTask.class;
+  }
+
+  @Override
+  protected List<Configuration> takeConfiguration(int maxTasks) {
+    return IntStream.range(0, maxTasks).mapToObj(i -> config).collect(Collectors.toList());
+  }
+
+  @Override
+  protected List<Definition> definitions() {
+    return List.of(FREQUENCY_DEF);
+  }
+}

connector/src/main/java/org/astraea/connector/perf/PerfSinkTask.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.connector.perf;
+
+import java.time.Duration;
+import java.util.List;
+import org.astraea.common.Configuration;
+import org.astraea.common.Utils;
+import org.astraea.common.consumer.Record;
+import org.astraea.connector.SinkTask;
+
+public class PerfSinkTask extends SinkTask {
+
+  private Duration frequency = Utils.toDuration(PerfSink.FREQUENCY_DEF.defaultValue().toString());
+
+  private volatile long lastPut = System.currentTimeMillis();
+
+  @Override
+  protected void init(Configuration configuration) {
+    frequency =
+        configuration
+            .string(PerfSource.FREQUENCY_DEF.name())
+            .map(Utils::toDuration)
+            .orElse(frequency);
+  }
+
+  @Override
+  protected void put(List<Record<byte[], byte[]>> records) {
+    var now = System.currentTimeMillis();
+    var diff = frequency.toMillis() - (now - lastPut);
+    if (diff > 0) Utils.sleep(Duration.ofMillis(diff));
+    lastPut = now;
+  }
+}

connector/src/main/java/org/astraea/connector/perf/PerfConnector.java → connector/src/main/java/org/astraea/connector/perf/PerfSource.java

@@ -25,7 +25,7 @@
 import org.astraea.connector.SourceConnector;
 import org.astraea.connector.SourceTask;
 
-public class PerfConnector extends SourceConnector {
+public class PerfSource extends SourceConnector {
   static Definition FREQUENCY_DEF =
       Definition.builder()
           .name("frequency.in.seconds")
@@ -47,7 +47,7 @@ protected void init(Configuration configuration) {
 
   @Override
   protected Class<? extends SourceTask> task() {
-    return PerfTask.class;
+    return PerfSourceTask.class;
   }
 
   @Override

connector/src/main/java/org/astraea/connector/perf/PerfTask.java → connector/src/main/java/org/astraea/connector/perf/PerfSourceTask.java

@@ -28,31 +28,25 @@
 import org.astraea.connector.SourceConnector;
 import org.astraea.connector.SourceTask;
 
-public class PerfTask extends SourceTask {
+public class PerfSourceTask extends SourceTask {
 
   private Set<String> topics = Set.of();
-  private int keyLength = (int) PerfConnector.KEY_LENGTH_DEF.defaultValue();
-  private int valueLength = (int) PerfConnector.KEY_LENGTH_DEF.defaultValue();
-  private Duration frequency =
-      Utils.toDuration(PerfConnector.FREQUENCY_DEF.defaultValue().toString());
+  private int keyLength = (int) PerfSource.KEY_LENGTH_DEF.defaultValue();
+  private int valueLength = (int) PerfSource.KEY_LENGTH_DEF.defaultValue();
+  private Duration frequency = Utils.toDuration(PerfSource.FREQUENCY_DEF.defaultValue().toString());
   private long last = System.currentTimeMillis();
 
   @Override
   protected void init(Configuration configuration) {
     this.topics = Set.copyOf(configuration.list(SourceConnector.TOPICS_KEY, ","));
-    this.keyLength =
-        configuration
-            .integer(PerfConnector.KEY_LENGTH_DEF.name())
-            .orElse((Integer) PerfConnector.KEY_LENGTH_DEF.defaultValue());
+    this.keyLength = configuration.integer(PerfSource.KEY_LENGTH_DEF.name()).orElse(keyLength);
     this.valueLength =
-        configuration
-            .integer(PerfConnector.VALUE_LENGTH_DEF.name())
-            .orElse((Integer) PerfConnector.VALUE_LENGTH_DEF.defaultValue());
+        configuration.integer(PerfSource.VALUE_LENGTH_DEF.name()).orElse(valueLength);
     this.frequency =
-        Utils.toDuration(
-            configuration
-                .string(PerfConnector.FREQUENCY_DEF.name())
-                .orElse((String) PerfConnector.FREQUENCY_DEF.defaultValue()));
+        configuration
+            .string(PerfSource.FREQUENCY_DEF.name())
+            .map(Utils::toDuration)
+            .orElse(frequency);
   }
 
   @Override

connector/src/test/java/org/astraea/connector/perf/PerfSinkTest.java

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.connector.perf;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.astraea.common.Configuration;
+import org.astraea.common.Utils;
+import org.astraea.common.connector.ConnectorClient;
+import org.astraea.common.consumer.Record;
+import org.astraea.it.RequireSingleWorkerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PerfSinkTest extends RequireSingleWorkerCluster {
+
+  @Test
+  void testDefaultConfig() {
+    var client = ConnectorClient.builder().url(workerUrl()).build();
+    var validation =
+        client
+            .validate(
+                PerfSink.class.getSimpleName(),
+                Map.of(
+                    ConnectorClient.NAME_KEY,
+                    Utils.randomString(),
+                    ConnectorClient.CONNECTOR_CLASS_KEY,
+                    PerfSink.class.getName(),
+                    ConnectorClient.TASK_MAX_KEY,
+                    "1",
+                    ConnectorClient.TOPICS_KEY,
+                    "abc"))
+            .toCompletableFuture()
+            .join();
+    Assertions.assertEquals(
+        0,
+        validation.errorCount(),
+        validation.configs().stream()
+            .flatMap(c -> c.value().errors().stream())
+            .collect(Collectors.joining(",")));
+  }
+
+  @Test
+  void testFrequency() {
+    var client = ConnectorClient.builder().url(workerUrl()).build();
+    var validation =
+        client
+            .validate(
+                PerfSink.class.getSimpleName(),
+                Map.of(
+                    ConnectorClient.NAME_KEY,
+                    Utils.randomString(),
+                    ConnectorClient.CONNECTOR_CLASS_KEY,
+                    PerfSink.class.getName(),
+                    ConnectorClient.TASK_MAX_KEY,
+                    "1",
+                    ConnectorClient.TOPICS_KEY,
+                    "abc",
+                    PerfSink.FREQUENCY_DEF.name(),
+                    "a"))
+            .toCompletableFuture()
+            .join();
+    Assertions.assertEquals(1, validation.errorCount());
+    Assertions.assertNotEquals(
+        0,
+        validation.configs().stream()
+            .filter(c -> c.definition().name().equals(PerfSink.FREQUENCY_DEF.name()))
+            .findFirst()
+            .get()
+            .value()
+            .errors()
+            .size());
+  }
+
+  @Test
+  void testTask() {
+    var task = new PerfSinkTask();
+    task.init(Configuration.of(Map.of(PerfSink.FREQUENCY_DEF.name(), "1s")));
+
+    var now = System.currentTimeMillis();
+    task.put(List.<Record<byte[], byte[]>>of());
+
+    // the frequency is 1s so the elapsed time of executing put should be greater than 500ms
+    Assertions.assertTrue(System.currentTimeMillis() - now > 500);
+  }
+}