You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by ta...@apache.org on 2021/08/16 09:10:39 UTC
[incubator-nemo] branch master updated: [NEMO-482] Keep library
versions updated (#316)
This is an automated email from the ASF dual-hosted git repository.
taegeonum pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
The following commit(s) were added to refs/heads/master by this push:
new e829c3c [NEMO-482] Keep library versions updated (#316)
e829c3c is described below
commit e829c3cbcf4a9fe395faf18f4b2e298231725055
Author: Won Wook SONG <ws...@gmail.com>
AuthorDate: Mon Aug 16 18:10:31 2021 +0900
[NEMO-482] Keep library versions updated (#316)
JIRA: [NEMO-482: Keep library versions updated](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-482)
**Major changes:**
- Updates library versions, and performs migrations for the version updates
**Minor changes to note:**
- NEXMark pom was updated to keep up with the changes
**Tests for the changes:**
- Existing tests comply with the changes
- NEXMark now also has an ITCase test
**Other comments:**
- None
Closes #316
---
README.md | 2 +-
bin/run_nexmark.sh | 2 +-
.../java/org/apache/nemo/client/ClientUtils.java | 16 +++++-
common/pom.xml | 5 ++
.../examples/beam/WindowedBroadcastITCase.java | 3 +-
.../examples/beam/WindowedWordCountITCase.java | 7 ++-
examples/nexmark/pom.xml | 35 +++++++++++-
.../nemo/examples/nexmark/NexmarkITCase.java | 65 ++++++++++++++++++++++
pom.xml | 51 ++++++++---------
.../nemo/runtime/master/PlanStateManager.java | 43 +++++++-------
10 files changed, 174 insertions(+), 55 deletions(-)
diff --git a/README.md b/README.md
index fd61c4b..0ff7859 100644
--- a/README.md
+++ b/README.md
@@ -121,7 +121,7 @@ $ ./bin/run_nexmark.sh \
-user_main org.apache.beam.sdk.nexmark.Main \
-optimization_policy org.apache.nemo.compiler.optimizer.policy.StreamingPolicy \
-scheduler_impl_class_name org.apache.nemo.runtime.master.scheduler.StreamingScheduler \
- -user_args "--runner=NemoRunner --streaming=true --query=0 --numEventGenerators=1"
+ -user_args "--runner=NemoRunner --streaming=true --query=0 --numEventGenerators=1 --manageResources=false --monitorJobs=false"
```
## Resource Configuration
diff --git a/bin/run_nexmark.sh b/bin/run_nexmark.sh
index ce2c0ef..10f40e7 100755
--- a/bin/run_nexmark.sh
+++ b/bin/run_nexmark.sh
@@ -21,4 +21,4 @@ VERSION=$(mvn -q \
-Dexec.executable=echo -Dexec.args='${project.version}' \
--non-recursive exec:exec)
-java -Dlog4j.configuration=file://`pwd`/log4j.properties -cp examples/nexmark/target/nexmark-${VERSION}-shaded.jar:client/target/nemo-client-${VERSION}-shaded.jar:`yarn classpath` org.apache.nemo.client.JobLauncher "$@"
+java -Dlog4j.configuration=file://`pwd`/log4j.properties -cp examples/nexmark/target/nexmark-${VERSION}-shaded.jar:client/target/nemo-client-${VERSION}-shaded.jar:`$YARN_HOME/bin/yarn classpath` org.apache.nemo.client.JobLauncher "$@"
diff --git a/client/src/main/java/org/apache/nemo/client/ClientUtils.java b/client/src/main/java/org/apache/nemo/client/ClientUtils.java
index afe9097..d9ad08d 100644
--- a/client/src/main/java/org/apache/nemo/client/ClientUtils.java
+++ b/client/src/main/java/org/apache/nemo/client/ClientUtils.java
@@ -19,7 +19,6 @@
package org.apache.nemo.client;
-import org.apache.commons.io.FileUtils;
import org.apache.nemo.common.Util;
import org.apache.nemo.common.exception.MetricException;
import org.apache.nemo.runtime.common.comm.ControlMessage;
@@ -27,6 +26,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.stream.Stream;
/**
* Utility class for the Client.
@@ -44,6 +48,7 @@ public final class ClientUtils {
* Handler for the launch optimization message.
*
* @param message the message received from the driver.
+ * @param driverRPCServer driver RPC server.
*/
static void handleOptimizationType(final ControlMessage.DriverToClientMessage message,
final DriverRPCServer driverRPCServer) {
@@ -83,8 +88,13 @@ public final class ClientUtils {
final String resultsFile = projectRootPath + "/ml/results.out";
LOG.info("Reading the results of the script at {}", resultsFile);
- return FileUtils.readFileToString(new File(resultsFile), "UTF-8");
- } catch (Exception e) {
+ final StringBuilder contentBuilder = new StringBuilder();
+ try (Stream<String> stream = Files.lines(Paths.get(resultsFile), StandardCharsets.UTF_8)) {
+ stream.forEach(s -> contentBuilder.append(s).append("\n"));
+ }
+ return contentBuilder.toString();
+ } catch (IOException | InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new MetricException(e);
}
}
diff --git a/common/pom.xml b/common/pom.xml
index 3c27023..a0ab9ba 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -37,6 +37,11 @@ under the License.
<version>${reef.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons-lang.version}</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.reef</groupId>
<artifactId>reef-common</artifactId>
<version>${reef.version}</version>
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedBroadcastITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedBroadcastITCase.java
index 06a72f4..daf7a9c 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedBroadcastITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedBroadcastITCase.java
@@ -22,6 +22,7 @@ import org.apache.nemo.client.JobLauncher;
import org.apache.nemo.common.test.ArgBuilder;
import org.apache.nemo.common.test.ExampleTestUtil;
import org.apache.nemo.examples.beam.policy.StreamingPolicyParallelismFive;
+import org.apache.nemo.runtime.master.scheduler.StreamingScheduler;
import org.junit.runner.RunWith;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
@@ -48,7 +49,7 @@ public final class WindowedBroadcastITCase {
// @Test (timeout = TIMEOUT)
public void testUnboundedSlidingWindow() throws Exception {
builder = new ArgBuilder()
- .addScheduler("org.apache.nemo.runtime.master.scheduler.StreamingScheduler")
+ .addScheduler(StreamingScheduler.class.getCanonicalName())
.addUserMain(WindowedBroadcast.class.getCanonicalName())
.addUserArgs(outputFilePath);
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedWordCountITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedWordCountITCase.java
index 3628f7c..67c63a1 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedWordCountITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/WindowedWordCountITCase.java
@@ -25,6 +25,7 @@ import org.apache.nemo.common.test.ExampleTestUtil;
import org.apache.nemo.compiler.optimizer.policy.DefaultPolicy;
import org.apache.nemo.examples.beam.policy.DefaultPolicyParallelismFive;
import org.apache.nemo.examples.beam.policy.StreamingPolicyParallelismFive;
+import org.apache.nemo.runtime.master.scheduler.StreamingScheduler;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.core.classloader.annotations.PrepareForTest;
@@ -92,7 +93,7 @@ public final class WindowedWordCountITCase {
@Test(timeout = ExampleTestArgs.TIMEOUT)
public void testStreamingSchedulerAndPipeFixedWindow() throws Exception {
builder = new ArgBuilder()
- .addScheduler("org.apache.nemo.runtime.master.scheduler.StreamingScheduler")
+ .addScheduler(StreamingScheduler.class.getCanonicalName())
.addUserMain(WindowedWordCount.class.getCanonicalName())
.addUserArgs(outputFilePath, "fixed", INPUT_TYPE_BOUNDED, inputFilePath);
@@ -112,7 +113,7 @@ public final class WindowedWordCountITCase {
@Test(timeout = ExampleTestArgs.TIMEOUT)
public void testStreamingSchedulerAndPipeSlidingWindow() throws Exception {
builder = new ArgBuilder()
- .addScheduler("org.apache.nemo.runtime.master.scheduler.StreamingScheduler")
+ .addScheduler(StreamingScheduler.class.getCanonicalName())
.addUserMain(WindowedWordCount.class.getCanonicalName())
.addUserArgs(outputFilePath, "sliding", INPUT_TYPE_BOUNDED, inputFilePath);
@@ -133,7 +134,7 @@ public final class WindowedWordCountITCase {
// @Test (timeout = TIMEOUT)
public void testUnboundedSlidingWindow() throws Exception {
builder = new ArgBuilder()
- .addScheduler("org.apache.nemo.runtime.master.scheduler.StreamingScheduler")
+ .addScheduler(StreamingScheduler.class.getCanonicalName())
.addUserMain(WindowedWordCount.class.getCanonicalName())
.addUserArgs(outputFilePath, "sliding", INPUT_TYPE_UNBOUNDED);
diff --git a/examples/nexmark/pom.xml b/examples/nexmark/pom.xml
index 9d6557b..71672bf 100644
--- a/examples/nexmark/pom.xml
+++ b/examples/nexmark/pom.xml
@@ -25,7 +25,7 @@ under the License.
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>nexmark</artifactId>
+ <artifactId>nemo-examples-nexmark</artifactId>
<name>Nemo Examples: NEXMark</name>
<dependencies>
@@ -41,6 +41,29 @@ under the License.
<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
<version>${beam.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.nemo</groupId>
+ <artifactId>nemo-client</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
<build>
@@ -56,6 +79,16 @@ under the License.
<goal>shade</goal>
</goals>
<configuration>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
<outputFile>
${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar
</outputFile>
diff --git a/examples/nexmark/src/test/java/org/apache/nemo/examples/nexmark/NexmarkITCase.java b/examples/nexmark/src/test/java/org/apache/nemo/examples/nexmark/NexmarkITCase.java
new file mode 100644
index 0000000..be4c2ea
--- /dev/null
+++ b/examples/nexmark/src/test/java/org/apache/nemo/examples/nexmark/NexmarkITCase.java
@@ -0,0 +1,65 @@
+/*
+ * 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.nemo.examples.nexmark;
+
+import org.apache.beam.sdk.nexmark.Main;
+import org.apache.nemo.client.JobLauncher;
+import org.apache.nemo.common.test.ArgBuilder;
+import org.apache.nemo.common.test.ExampleTestArgs;
+import org.apache.nemo.compiler.optimizer.policy.StreamingPolicy;
+import org.apache.nemo.runtime.master.scheduler.StreamingScheduler;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+/**
+ * Test Nexmark execution with JobLauncher.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(JobLauncher.class)
+public class NexmarkITCase {
+ private static ArgBuilder builder;
+
+ private static final String executorResourceFileName = ExampleTestArgs.getFileBasePath() + "executors/beam_test_executor_resources.json";
+
+ @Before
+ public void setUp() throws Exception {
+ builder = new ArgBuilder()
+ .addResourceJson(executorResourceFileName)
+ .addUserMain(Main.class.getCanonicalName())
+ .addOptimizationPolicy(StreamingPolicy.class.getCanonicalName())
+ .addScheduler(StreamingScheduler.class.getCanonicalName());
+ }
+
+ /**
+ * Testing query 0.
+ * @throws Exception exception.
+ */
+ @Test(timeout = ExampleTestArgs.TIMEOUT)
+ public void testQuery0() throws Exception {
+ JobLauncher.main(builder
+ .addJobId("Nexmark-Q0")
+ .addUserArgs("--runner=NemoRunner", "--streaming=false", "--query=0",
+ "--numEventGenerators=1", "--manageResources=false", "--monitorJobs=false")
+ .build());
+ }
+}
diff --git a/pom.xml b/pom.xml
index 834ea5f..f26bf58 100644
--- a/pom.xml
+++ b/pom.xml
@@ -44,36 +44,37 @@ under the License.
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
<maven.compiler.version>3.8.1</maven.compiler.version>
- <maven.shade.version>3.2.1</maven.shade.version>
+ <maven.shade.version>3.2.4</maven.shade.version>
<maven.deploy.version>3.0.0-M1</maven.deploy.version>
- <maven.javadoc.version>3.1.1</maven.javadoc.version>
+ <maven.javadoc.version>3.3.0</maven.javadoc.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <auto-service.version>1.0-rc2</auto-service.version>
- <beam.version>2.18.0</beam.version>
+ <auto-service.version>1.0</auto-service.version>
+ <beam.version>2.24.0</beam.version> <!-- > 2.25 needs migration -->
<spark.version>2.4.4</spark.version>
<scala.version>2.12.10</scala.version>
<kryo.version>4.0.2</kryo.version>
<reef.version>0.16.0</reef.version>
- <protobuf.version>2.5.0</protobuf.version>
+ <protobuf.version>3.17.3</protobuf.version>
<hadoop.version>2.7.2</hadoop.version>
<log4j.configuration>file://log4j.properties</log4j.configuration>
- <netty.version>4.1.42.Final</netty.version>
- <guava.version>22.0</guava.version>
- <grpc.version>1.26.0</grpc.version>
- <jackson.version>2.10.2</jackson.version>
+ <netty.version>4.1.66.Final</netty.version>
+ <guava.version>23.0</guava.version>
+ <grpc.version>1.39.0</grpc.version>
+ <jackson.version>2.10.5</jackson.version> <!-- > 2.11 needs migration -->
<netlib.version>1.1.2</netlib.version>
- <jetty-server.version>9.4.10.v20180503</jetty-server.version>
- <jetty-servlet.version>9.4.10.v20180503</jetty-servlet.version>
+ <jetty-server.version>9.4.43.v20210629</jetty-server.version>
+ <jetty-servlet.version>9.4.43.v20210629</jetty-servlet.version>
<commons-math.version>3.6.1</commons-math.version>
- <sqlite-jdbc.version>3.30.1</sqlite-jdbc.version>
- <postgresql.version>42.2.9</postgresql.version>
- <slf4j.version>1.7.30</slf4j.version>
+ <commons-lang.version>3.12.0</commons-lang.version>
+ <sqlite-jdbc.version>3.36.0</sqlite-jdbc.version>
+ <postgresql.version>42.2.23</postgresql.version>
+ <slf4j.version>1.7.32</slf4j.version>
<!-- Tests -->
- <mockito.version>3.2.4</mockito.version>
- <powermock.version>2.0.4</powermock.version>
- <surefire.version>3.0.0-M4</surefire.version>
- <junit.version>4.13</junit.version>
- <checkstyle.version>8.16</checkstyle.version>
+ <mockito.version>3.11.2</mockito.version>
+ <powermock.version>2.0.9</powermock.version>
+ <surefire.version>3.0.0-M4</surefire.version> <!-- > M5 errors with Java11 -->
+ <junit.version>4.13.2</junit.version>
+ <checkstyle.version>8.23</checkstyle.version>
</properties>
<modules>
@@ -94,7 +95,7 @@ under the License.
<dependency>
<groupId>org.xerial.snappy</groupId>
<artifactId>snappy-java</artifactId>
- <version>1.1.1.3</version>
+ <version>1.1.8.4</version>
</dependency>
<!--tests-->
<dependency>
@@ -165,7 +166,7 @@ under the License.
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
- <version>1.5.0.Final</version>
+ <version>1.7.0</version>
</extension>
</extensions>
@@ -200,7 +201,7 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
- <version>3.0.0</version>
+ <version>3.1.2</version>
<dependencies>
<dependency>
<groupId>com.puppycrawl.tools</groupId>
@@ -243,7 +244,7 @@ under the License.
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
- <version>3.0.0</version>
+ <version>3.2.0</version>
<executions>
<execution>
<id>add-source</id>
@@ -262,7 +263,7 @@ under the License.
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
- <version>0.5.0</version>
+ <version>0.6.1</version>
<configuration>
<pluginId>grpc-java</pluginId>
<protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
@@ -300,7 +301,7 @@ under the License.
<plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
- <version>0.12</version>
+ <version>0.13</version>
<executions>
<execution>
<phase>verify</phase>
diff --git a/runtime/master/src/main/java/org/apache/nemo/runtime/master/PlanStateManager.java b/runtime/master/src/main/java/org/apache/nemo/runtime/master/PlanStateManager.java
index 53cab57..fd02da9 100644
--- a/runtime/master/src/main/java/org/apache/nemo/runtime/master/PlanStateManager.java
+++ b/runtime/master/src/main/java/org/apache/nemo/runtime/master/PlanStateManager.java
@@ -633,30 +633,33 @@ public final class PlanStateManager {
public synchronized String toString() {
final StringBuilder sb = new StringBuilder("{");
sb.append("\"planId\": \"").append(planId).append("\", ");
- sb.append("\"stages\": [");
- boolean isFirstStage = true;
- for (final Stage stage : physicalPlan.getStageDAG().getVertices()) {
- if (!isFirstStage) {
- sb.append(", ");
- }
- isFirstStage = false;
- final StageState stageState = stageIdToState.get(stage.getId());
- sb.append("{\"id\": \"").append(stage.getId()).append("\", ");
- sb.append("\"state\": \"").append(stageState.toString()).append("\", ");
- sb.append("\"tasks\": [");
-
- boolean isFirstTask = true;
- for (final Map.Entry<String, TaskState.State> entry : getTaskAttemptIdsToItsState(stage.getId()).entrySet()) {
- if (!isFirstTask) {
+ if (physicalPlan != null) {
+ sb.append("\"stages\": [");
+ boolean isFirstStage = true;
+ for (final Stage stage : physicalPlan.getStageDAG().getVertices()) {
+ if (!isFirstStage) {
sb.append(", ");
}
- isFirstTask = false;
- sb.append("{\"id\": \"").append(entry.getKey()).append("\", ");
- sb.append("\"state\": \"").append(entry.getValue().toString()).append("\"}");
+ isFirstStage = false;
+ final StageState stageState = stageIdToState.get(stage.getId());
+ sb.append("{\"id\": \"").append(stage.getId()).append("\", ");
+ sb.append("\"state\": \"").append(stageState.toString()).append("\", ");
+ sb.append("\"tasks\": [");
+
+ boolean isFirstTask = true;
+ for (final Map.Entry<String, TaskState.State> entry : getTaskAttemptIdsToItsState(stage.getId()).entrySet()) {
+ if (!isFirstTask) {
+ sb.append(", ");
+ }
+ isFirstTask = false;
+ sb.append("{\"id\": \"").append(entry.getKey()).append("\", ");
+ sb.append("\"state\": \"").append(entry.getValue().toString()).append("\"}");
+ }
+ sb.append("]}");
}
- sb.append("]}");
+ sb.append("]");
}
- sb.append("]}");
+ sb.append("}");
return sb.toString();
}
}