You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2020/03/16 06:50:06 UTC
[flink-statefun] branch master updated (3477f45 -> 0588839)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git.
from 3477f45 [FLINK-16558] [doc] Reword project website tagline
new e4f847e [FLINK-16254] Add parallelism to StatefulFunctionsClusterConfiguration
new c7e5b26 [FLINK-16254] Use parallelism config to create Stateful Functions JobGraph
new 0588839 [FLINK-16254] [e2e] Adapt StatefulFunctionsAppContainers to use the -p command option
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../e2e/common/StatefulFunctionsAppContainers.java | 11 ++++-------
.../StatefulFunctionsClusterConfiguration.java | 10 +++++++++-
...FunctionsClusterConfigurationParserFactory.java | 23 +++++++++++++++++++++-
.../StatefulFunctionsClusterEntryPoint.java | 8 +++++++-
.../StatefulFunctionsJobGraphRetriever.java | 23 +++++++++++++++++++---
5 files changed, 62 insertions(+), 13 deletions(-)
[flink-statefun] 01/03: [FLINK-16254] Add parallelism to
StatefulFunctionsClusterConfiguration
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit e4f847ef9a060f0caf87ee20fec02f1df4dbfec9
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Feb 24 15:41:31 2020 +0800
[FLINK-16254] Add parallelism to StatefulFunctionsClusterConfiguration
This commit adds parallelism value to
StatefulFunctionsClusterConfiguration, parsed from the command line.
---
.../StatefulFunctionsClusterConfiguration.java | 10 +++++++++-
...FunctionsClusterConfigurationParserFactory.java | 23 +++++++++++++++++++++-
2 files changed, 31 insertions(+), 2 deletions(-)
diff --git a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfiguration.java b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfiguration.java
index d2ff55f..0afcfea 100644
--- a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfiguration.java
+++ b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfiguration.java
@@ -37,6 +37,8 @@ final class StatefulFunctionsClusterConfiguration extends EntrypointClusterConfi
@Nullable private final JobID jobId;
+ private final int parallelism;
+
StatefulFunctionsClusterConfiguration(
@Nonnull String configDir,
@Nonnull Properties dynamicProperties,
@@ -44,11 +46,13 @@ final class StatefulFunctionsClusterConfiguration extends EntrypointClusterConfi
@Nullable String hostname,
int restPort,
@Nonnull SavepointRestoreSettings savepointRestoreSettings,
- @Nullable JobID jobId) {
+ @Nullable JobID jobId,
+ int parallelism) {
super(configDir, dynamicProperties, args, hostname, restPort);
this.savepointRestoreSettings =
requireNonNull(savepointRestoreSettings, "savepointRestoreSettings");
this.jobId = jobId;
+ this.parallelism = parallelism;
}
@Nonnull
@@ -60,4 +64,8 @@ final class StatefulFunctionsClusterConfiguration extends EntrypointClusterConfi
JobID getJobId() {
return jobId;
}
+
+ public int getParallelism() {
+ return parallelism;
+ }
}
diff --git a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfigurationParserFactory.java b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfigurationParserFactory.java
index da6cbfb..6b00f4c 100644
--- a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfigurationParserFactory.java
+++ b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterConfigurationParserFactory.java
@@ -17,6 +17,7 @@
*/
package org.apache.flink.statefun.flink.launcher;
+import static org.apache.flink.client.cli.CliFrontendParser.PARALLELISM_OPTION;
import static org.apache.flink.runtime.entrypoint.parser.CommandLineOptions.CONFIG_DIR_OPTION;
import static org.apache.flink.runtime.entrypoint.parser.CommandLineOptions.DYNAMIC_PROPERTY_OPTION;
import static org.apache.flink.runtime.entrypoint.parser.CommandLineOptions.HOST_OPTION;
@@ -28,6 +29,7 @@ import javax.annotation.Nullable;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.cli.CliFrontendParser;
import org.apache.flink.runtime.entrypoint.FlinkParseException;
@@ -77,6 +79,7 @@ public class StatefulFunctionsClusterConfigurationParserFactory
options.addOption(REST_PORT_OPTION);
options.addOption(JOB_ID_OPTION);
options.addOption(DYNAMIC_PROPERTY_OPTION);
+ options.addOption(PARALLELISM_OPTION);
options.addOption(CliFrontendParser.SAVEPOINT_PATH_OPTION);
options.addOption(CliFrontendParser.SAVEPOINT_ALLOW_NON_RESTORED_OPTION);
@@ -91,6 +94,7 @@ public class StatefulFunctionsClusterConfigurationParserFactory
commandLine.getOptionProperties(DYNAMIC_PROPERTY_OPTION.getOpt());
final int restPort = getRestPort(commandLine);
final String hostname = commandLine.getOptionValue(HOST_OPTION.getOpt());
+ final int parallelism = getParallelism(commandLine);
final SavepointRestoreSettings savepointRestoreSettings =
CliFrontendParser.createSavepointRestoreSettings(commandLine);
final JobID jobId = getJobId(commandLine);
@@ -102,7 +106,8 @@ public class StatefulFunctionsClusterConfigurationParserFactory
hostname,
restPort,
savepointRestoreSettings,
- jobId);
+ jobId,
+ parallelism);
}
private int getRestPort(CommandLine commandLine) throws FlinkParseException {
@@ -113,4 +118,20 @@ public class StatefulFunctionsClusterConfigurationParserFactory
throw createFlinkParseException(REST_PORT_OPTION, e);
}
}
+
+ private int getParallelism(CommandLine commandLine) throws FlinkParseException {
+ final String parallelismString =
+ commandLine.getOptionValue(
+ PARALLELISM_OPTION.getOpt(), String.valueOf(ExecutionConfig.PARALLELISM_DEFAULT));
+ try {
+ int parallelism = Integer.parseInt(parallelismString);
+ if (parallelism <= 0 && parallelism != ExecutionConfig.PARALLELISM_DEFAULT) {
+ throw new IllegalArgumentException(
+ "Parallelism must be at least 1. Provided: " + parallelism);
+ }
+ return parallelism;
+ } catch (NumberFormatException e) {
+ throw createFlinkParseException(PARALLELISM_OPTION, e);
+ }
+ }
}
[flink-statefun] 03/03: [FLINK-16254] [e2e] Adapt
StatefulFunctionsAppContainers to use the -p command option
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit 05888397d049f23c9bc960a667d8350570c5f95c
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sat Mar 14 14:25:25 2020 +0800
[FLINK-16254] [e2e] Adapt StatefulFunctionsAppContainers to use the -p command option
This closes #58.
---
.../statefun/e2e/common/StatefulFunctionsAppContainers.java | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/statefun-e2e-tests/statefun-e2e-tests-common/src/main/java/org/apache/flink/statefun/e2e/common/StatefulFunctionsAppContainers.java b/statefun-e2e-tests/statefun-e2e-tests-common/src/main/java/org/apache/flink/statefun/e2e/common/StatefulFunctionsAppContainers.java
index 5d1cc87..f648fcd 100644
--- a/statefun-e2e-tests/statefun-e2e-tests-common/src/main/java/org/apache/flink/statefun/e2e/common/StatefulFunctionsAppContainers.java
+++ b/statefun-e2e-tests/statefun-e2e-tests-common/src/main/java/org/apache/flink/statefun/e2e/common/StatefulFunctionsAppContainers.java
@@ -31,7 +31,6 @@ import java.util.Objects;
import javax.annotation.Nullable;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.statefun.flink.core.StatefulFunctionsConfig;
import org.junit.rules.ExternalResource;
@@ -142,10 +141,6 @@ public final class StatefulFunctionsAppContainers extends ExternalResource {
this.network = Network.newNetwork();
this.appName = appName;
this.numWorkers = numWorkers;
-
- if (numWorkers > 1) {
- dynamicProperties.set(CoreOptions.DEFAULT_PARALLELISM, numWorkers);
- }
}
public StatefulFunctionsAppContainers dependsOn(GenericContainer<?> container) {
@@ -180,7 +175,7 @@ public final class StatefulFunctionsAppContainers extends ExternalResource {
protected void before() throws Throwable {
final ImageFromDockerfile appImage =
appImage(appName, dynamicProperties, classpathBuildContextFiles);
- this.master = masterContainer(appImage, network, dependentContainers, masterLogger);
+ this.master = masterContainer(appImage, network, dependentContainers, numWorkers, masterLogger);
this.workers = workerContainers(appImage, numWorkers, network);
master.start();
@@ -261,13 +256,15 @@ public final class StatefulFunctionsAppContainers extends ExternalResource {
ImageFromDockerfile appImage,
Network network,
List<GenericContainer<?>> dependents,
+ int numWorkers,
@Nullable Logger masterLogger) {
final GenericContainer<?> master =
new GenericContainer(appImage)
.withNetwork(network)
.withNetworkAliases(MASTER_HOST)
.withEnv("ROLE", "master")
- .withEnv("MASTER_HOST", MASTER_HOST);
+ .withEnv("MASTER_HOST", MASTER_HOST)
+ .withCommand("-p " + numWorkers);
for (GenericContainer<?> dependent : dependents) {
master.dependsOn(dependent);
[flink-statefun] 02/03: [FLINK-16254] Use parallelism config to
create Stateful Functions JobGraph
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit c7e5b26ef1cfc729da36bbe4e04b30ea7d14b040
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Mon Feb 24 15:50:10 2020 +0800
[FLINK-16254] Use parallelism config to create Stateful Functions JobGraph
---
.../StatefulFunctionsClusterEntryPoint.java | 8 +++++++-
.../StatefulFunctionsJobGraphRetriever.java | 23 +++++++++++++++++++---
2 files changed, 27 insertions(+), 4 deletions(-)
diff --git a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java
index be22830..65a73bd 100644
--- a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java
+++ b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsClusterEntryPoint.java
@@ -47,18 +47,22 @@ public final class StatefulFunctionsClusterEntryPoint extends JobClusterEntrypoi
@Nonnull private final SavepointRestoreSettings savepointRestoreSettings;
+ private final int parallelism;
+
@Nonnull private final String[] programArguments;
private StatefulFunctionsClusterEntryPoint(
Configuration configuration,
@Nonnull JobID jobId,
@Nonnull SavepointRestoreSettings savepointRestoreSettings,
+ int parallelism,
@Nonnull String[] programArguments) {
super(configuration);
this.jobId = requireNonNull(jobId, "jobId");
this.savepointRestoreSettings =
requireNonNull(savepointRestoreSettings, "savepointRestoreSettings");
this.programArguments = requireNonNull(programArguments, "programArguments");
+ this.parallelism = parallelism;
}
public static void main(String[] args) {
@@ -89,6 +93,7 @@ public final class StatefulFunctionsClusterEntryPoint extends JobClusterEntrypoi
resolveJobIdForCluster(
Optional.ofNullable(clusterConfiguration.getJobId()), configuration),
clusterConfiguration.getSavepointRestoreSettings(),
+ clusterConfiguration.getParallelism(),
clusterConfiguration.getArgs());
ClusterEntrypoint.runClusterEntrypoint(entrypoint);
@@ -141,6 +146,7 @@ public final class StatefulFunctionsClusterEntryPoint extends JobClusterEntrypoi
createDispatcherResourceManagerComponentFactory(Configuration configuration) {
return DefaultDispatcherResourceManagerComponentFactory.createJobComponentFactory(
StandaloneResourceManagerFactory.INSTANCE,
- new StatefulFunctionsJobGraphRetriever(jobId, savepointRestoreSettings, programArguments));
+ new StatefulFunctionsJobGraphRetriever(
+ jobId, savepointRestoreSettings, parallelism, programArguments));
}
}
diff --git a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsJobGraphRetriever.java b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsJobGraphRetriever.java
index 683baf4..5f7e642 100644
--- a/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsJobGraphRetriever.java
+++ b/statefun-flink/statefun-flink-launcher/src/main/java/org/apache/flink/statefun/flink/launcher/StatefulFunctionsJobGraphRetriever.java
@@ -25,6 +25,7 @@ import java.net.URI;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramUtils;
@@ -54,13 +55,18 @@ final class StatefulFunctionsJobGraphRetriever implements JobGraphRetriever {
private final JobID jobId;
private final SavepointRestoreSettings savepointRestoreSettings;
+ private final int parallelism;
private final String[] programArguments;
StatefulFunctionsJobGraphRetriever(
- JobID jobId, SavepointRestoreSettings savepointRestoreSettings, String[] programArguments) {
+ JobID jobId,
+ SavepointRestoreSettings savepointRestoreSettings,
+ int parallelism,
+ String[] programArguments) {
this.jobId = requireNonNull(jobId, "jobId");
this.savepointRestoreSettings =
requireNonNull(savepointRestoreSettings, "savepointRestoreSettings");
+ this.parallelism = parallelism;
this.programArguments = requireNonNull(programArguments, "programArguments");
}
@@ -84,11 +90,16 @@ final class StatefulFunctionsJobGraphRetriever implements JobGraphRetriever {
public JobGraph retrieveJobGraph(Configuration configuration) throws FlinkException {
final PackagedProgram packagedProgram = createPackagedProgram();
- final int defaultParallelism = configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM);
+ int resolvedParallelism = resolveParallelism(parallelism, configuration);
+ LOG.info(
+ "Creating JobGraph for job {}, with parallelism {} and savepoint restore settings {}.",
+ jobId,
+ resolvedParallelism,
+ savepointRestoreSettings);
try {
final JobGraph jobGraph =
PackagedProgramUtils.createJobGraph(
- packagedProgram, configuration, defaultParallelism, jobId, false);
+ packagedProgram, configuration, resolvedParallelism, jobId, false);
jobGraph.setSavepointRestoreSettings(savepointRestoreSettings);
return jobGraph;
@@ -113,4 +124,10 @@ final class StatefulFunctionsJobGraphRetriever implements JobGraphRetriever {
throw new RuntimeException("Unable to construct a packaged program", e);
}
}
+
+ private static int resolveParallelism(int parallelism, Configuration configuration) {
+ return (parallelism == ExecutionConfig.PARALLELISM_DEFAULT)
+ ? configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM)
+ : parallelism;
+ }
}