This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new e9b33751 [CELEBORN-289] Add flink integration test module. (#1229)
e9b33751 is described below
commit e9b33751d30cc15dbfd2936224f26be7d4f63a11
Author: Ethan Feng <[email protected]>
AuthorDate: Tue Feb 21 12:25:23 2023 +0800
[CELEBORN-289] Add flink integration test module. (#1229)
---
pom.xml | 16 +++
tests/flink-it/pom.xml | 119 +++++++++++++++++++++
.../celeborn/tests/flink/WordCountHelper.java | 103 ++++++++++++++++++
tests/flink-it/src/test/resources/log4j2-test.xml | 41 +++++++
.../celeborn/tests/flink/WordCountTest.scala | 74 +++++++++++++
5 files changed, 353 insertions(+)
diff --git a/pom.xml b/pom.xml
index cc6fcefd..2eee9009 100644
--- a/pom.xml
+++ b/pom.xml
@@ -166,6 +166,21 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
@@ -1037,6 +1052,7 @@
<module>client-flink/common</module>
<module>client-flink/flink-1.14</module>
<module>client-flink/flink-1.14-shaded</module>
+ <module>tests/flink-it</module>
</modules>
<properties>
<flink.version>1.14.6</flink.version>
diff --git a/tests/flink-it/pom.xml b/tests/flink-it/pom.xml
new file mode 100644
index 00000000..33be3e20
--- /dev/null
+++ b/tests/flink-it/pom.xml
@@ -0,0 +1,119 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-flink-it_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn Flink Integration Test</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-client_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-master_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-worker_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-worker_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-scala-scalatest_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>3.6.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-flink-1.14_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-scala_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git
a/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/WordCountHelper.java
b/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/WordCountHelper.java
new file mode 100644
index 00000000..bbb125d5
--- /dev/null
+++
b/tests/flink-it/src/test/java/org/apache/celeborn/tests/flink/WordCountHelper.java
@@ -0,0 +1,103 @@
+/*
+ * 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.apache.celeborn.tests.flink;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.util.Collector;
+
+public class WordCountHelper {
+
+ private static final int NUM_WORDS = 20;
+
+ private static final int WORD_COUNT = 200;
+
+ public static void execute(StreamExecutionEnvironment env, int parallelism) {
+ DataStream<Tuple2<String, Long>> words =
+ env.fromSequence(0, NUM_WORDS)
+ .broadcast()
+ .map(new WordsMapper())
+ .flatMap(new WordsFlatMapper(WORD_COUNT));
+ words
+ .keyBy(value -> value.f0)
+ .sum(1)
+ .map((MapFunction<Tuple2<String, Long>, Long>) wordCount ->
wordCount.f1)
+ .addSink(new VerifySink((long) parallelism * WORD_COUNT));
+ }
+
+ private static class WordsMapper implements MapFunction<Long, String> {
+
+ private static final long serialVersionUID = -896627105414186948L;
+
+ private static final String WORD_SUFFIX_1K = getWordSuffix1k();
+
+ private static String getWordSuffix1k() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("-");
+ for (int i = 0; i < 1024; ++i) {
+ builder.append("0");
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public String map(Long value) {
+ return "WORD-" + value + WORD_SUFFIX_1K;
+ }
+ }
+
+ private static class WordsFlatMapper implements FlatMapFunction<String,
Tuple2<String, Long>> {
+
+ private static final long serialVersionUID = 7873046672795114433L;
+
+ private final int wordsCount;
+
+ public WordsFlatMapper(int wordsCount) {
+ this.wordsCount = wordsCount;
+ }
+
+ @Override
+ public void flatMap(String word, Collector<Tuple2<String, Long>>
collector) {
+ for (int i = 0; i < wordsCount; ++i) {
+ collector.collect(new Tuple2<>(word, 1L));
+ }
+ }
+ }
+
+ private static class VerifySink implements SinkFunction<Long> {
+
+ private static final long serialVersionUID = -1975623991098131708L;
+
+ private final Long wordCount;
+
+ public VerifySink(long wordCount) {
+ this.wordCount = wordCount;
+ }
+
+ @Override
+ public void invoke(Long value, Context context) {
+ assertEquals(wordCount, value);
+ }
+ }
+}
diff --git a/tests/flink-it/src/test/resources/log4j2-test.xml
b/tests/flink-it/src/test/resources/log4j2-test.xml
new file mode 100644
index 00000000..5fb0cb8b
--- /dev/null
+++ b/tests/flink-it/src/test/resources/log4j2-test.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<Configuration status="INFO">
+ <Appenders>
+ <Console name="stdout" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{yy/MM/dd HH:mm:ss,SSS} %p [%t] %c{1}:
%m%n%ex"/>
+ <Filters>
+ <ThresholdFilter level="DEBUG"/>
+ </Filters>
+ </Console>
+ <File name="file" fileName="target/unit-tests.log">
+ <PatternLayout pattern="%d{yy/MM/dd HH:mm:ss,SSS} %p [%t] %c{1}:
%m%n%ex"/>
+ </File>
+ </Appenders>
+ <Loggers>
+ <Root level="INFO">
+ <AppenderRef ref="stdout"/>
+ <AppenderRef ref="file"/>
+ </Root>
+ <Logger name="org.sparkproject.jetty" level="WARN" additivity="false">
+ <AppenderRef ref="stdout"/>
+ <AppenderRef ref="file"/>
+ </Logger>
+ </Loggers>
+</Configuration>
diff --git
a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
new file mode 100644
index 00000000..14d247ba
--- /dev/null
+++
b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.apache.celeborn.tests.flink;
+
+import org.apache.flink.api.common.{ExecutionMode, InputDependencyConstraint,
RuntimeExecutionMode}
+import org.apache.flink.configuration.{ConfigConstants, Configuration,
ExecutionOptions, RestOptions}
+import org.apache.flink.runtime.jobgraph.JobType
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.service.deploy.MiniClusterFeature;
+
+class WordCountTest extends AnyFunSuite with Logging with MiniClusterFeature
+ with BeforeAndAfterAll {
+
+ override def beforeAll(): Unit = {
+ logInfo("test initialized , setup rss mini cluster")
+ val masterConf = Map(
+ "celeborn.master.host" -> "localhost",
+ "celeborn.master.port" -> "9097")
+ val workerConf = Map("celeborn.master.endpoints" -> "localhost:9097")
+ setUpMiniCluster(masterConf, workerConf)
+ }
+
+ override def afterAll(): Unit = {
+ logInfo("all test complete , stop rss mini cluster")
+ shutdownMiniCluster()
+ }
+
+ test("celeborn flink integration test - word count") {
+ // set up execution environment
+ val configuration = new Configuration
+ val parallelism = 8
+ configuration.setString(
+ "shuffle-service-factory.class",
+ "org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory")
+ configuration.setString("celeborn.master.endpoints", "localhost:9097")
+ configuration.setString("execution.batch-shuffle-mode",
"ALL_EXCHANGES_BLOCKING")
+ configuration.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true)
+ configuration.set(ExecutionOptions.RUNTIME_MODE,
RuntimeExecutionMode.BATCH)
+ configuration.setString("taskmanager.memory.network.min", "1024m")
+ configuration.setString(RestOptions.BIND_PORT, "8081-8089")
+ val env =
StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration)
+ env.getConfig.setExecutionMode(ExecutionMode.BATCH)
+ env.getConfig.setParallelism(parallelism)
+
env.getConfig.setDefaultInputDependencyConstraint(InputDependencyConstraint.ALL)
+ env.disableOperatorChaining()
+ // make parameters available in the web interface
+ WordCountHelper.execute(env, parallelism)
+
+ val graph = env.getStreamGraph
+
graph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_BLOCKING)
+ graph.setJobType(JobType.BATCH)
+ env.execute(graph)
+ }
+}
