You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tillrohrmann <gi...@git.apache.org> on 2018/07/11 23:20:37 UTC
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
GitHub user tillrohrmann opened a pull request:
https://github.com/apache/flink/pull/6320
[FLINK-9823] Add Kubernetes deployment ymls
## What is the purpose of the change
The Kubernetes files contain a job-cluster service specification, a job specification
for the StandaloneJobClusterEntryPoint and a deployment for TaskManagers.
This PR is based on #6319.
cc @GJL
## Verifying this change
- Tested manually
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (README)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tillrohrmann/flink containerEntrypoint
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6320.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6320
----
commit b798af824433657ca215f8598112094a23819ee0
Author: Till Rohrmann <tr...@...>
Date: 2018-07-11T15:30:53Z
[hotfix] Make PackagedProgram(Class<?>, String...) constructor public
commit d373b6b01ec9d5b63513718a8e6b7db87629a477
Author: Till Rohrmann <tr...@...>
Date: 2018-07-11T15:41:27Z
[FLINK-9818] Add cluster component command line parser
The cluster component command line parser is responsible for parsing the common command line
arguments with which the cluster components are started. These include the configDir, webui-port
and dynamic properties.
commit 2526bcc69ff4eb3d196a4a7ceba2e59a7f455922
Author: Till Rohrmann <tr...@...>
Date: 2018-07-09T21:54:55Z
[FLINK-9488] Add container entry point StandaloneJobClusterEntryPoint
The StandaloneJobClusterEntryPoint is the basic entry point for containers. It is started with
the user code jar in its classpath and the classname of the user program. The entrypoint will
then load this user program via the classname and execute its main method. This will generate
a JobGraph which is then used to start the MiniDispatcher.
commit 3b78e4099de1511bbe52c203fd2d05e5cfa03efa
Author: Till Rohrmann <tr...@...>
Date: 2018-07-10T09:24:26Z
[FLINK-9819] Add startup scripts for standalone job cluster entry point
commit 75fb8125e3ec270994628f46457b281cdb587874
Author: Till Rohrmann <tr...@...>
Date: 2018-07-10T21:23:59Z
[FLINK-9820] Forward dynamic properties to Flink configuration in ClusterEntrypoint
With this commit we can use dynamic properties to overwrite configuration values in the
ClusterEntrypoint.
commit b38683205961c625e8c99eff1552ef5a8142ee89
Author: Till Rohrmann <tr...@...>
Date: 2018-07-10T21:43:34Z
[FLINK-9821] Forward dynamic properties to configuration in TaskManagerRunner
With this commit we can use dynamic properties to overwrite configuration values in the
TaskManagerRunner.
commit 339a24fb2508c7f3cd041bc2cf9b15fd62980fcf
Author: Till Rohrmann <tr...@...>
Date: 2018-07-10T13:41:18Z
[FLINK-9822] Add Dockerfile for StandaloneJobClusterEntryPoint image
This commit adds a Dockerfile for a standalone job cluster image. The image
contains the Flink distribution and a specified user code jar. The entrypoint
will start the StandaloneJobClusterEntryPoint with the provided job classname.
commit c0f8ce88a1e5ce877add31214d9b2674acfbc90f
Author: Till Rohrmann <tr...@...>
Date: 2018-07-10T22:52:08Z
[FLINK-9823] Add Kubernetes deployment ymls
The Kubernetes files contain a job-cluster service specification, a job specification
for the StandaloneJobClusterEntryPoint and a deployment for TaskManagers.
commit b32a8f4149fecf953d24bd62af56f8620b360610
Author: Till Rohrmann <tr...@...>
Date: 2018-07-11T14:13:30Z
[hotfix] Support building a job image from a Flink archive
Extend the flink-container/docker/build.sh script to also accept a Flink archive to build
the image from. This makes it easier to build an image from one of the convenience releases.
----
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202376520
--- Diff: flink-container/pom.xml ---
@@ -0,0 +1,67 @@
+<!--
--- End diff --
Good catch. It slipped through. Will correct it.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202390083
--- Diff: flink-container/docker/build.sh ---
@@ -0,0 +1,128 @@
+#!/bin/sh
+
+################################################################################
+# 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.
+################################################################################
+
+usage() {
+ cat <<HERE
+Usage:
+ build.sh --job-jar <path-to-job-jar> --from-local-dist [--image-name <image>]
+ build.sh --job-jar <path-to-job-jar> --from-archive <path-to-dist-archive> [--image-name <image>]
+ build.sh --job-jar <path-to-job-jar> --from-release --flink-version <x.x.x> --hadoop-version <x.x> --scala-version <x.xx> [--image-name <image>]
+ build.sh --help
+
+ If the --image-name flag is not used the built image name will be 'flink'.
+HERE
+ exit 1
+}
+
+while [[ $# -ge 1 ]]
+do
+key="$1"
+ case $key in
+ --job-jar)
+ JOB_JAR_PATH="$2"
+ shift
+ ;;
+ --from-local-dist)
+ FROM_LOCAL="true"
+ ;;
+ --from-archive)
+ FROM_ARCHIVE="$2"
+ shift
+ ;;
+ --from-release)
+ FROM_RELEASE="true"
+ ;;
+ --image-name)
+ IMAGE_NAME="$2"
+ shift
+ ;;
+ --flink-version)
+ FLINK_VERSION="$2"
+ shift
+ ;;
+ --hadoop-version)
+ HADOOP_VERSION="$(echo "$2" | sed 's/\.//')"
+ shift
+ ;;
+ --scala-version)
+ SCALA_VERSION="$2"
+ shift
+ ;;
+ --kubernetes-certificates)
+ CERTIFICATES_DIR="$2"
+ shift
+ ;;
+ --help)
+ usage
+ ;;
+ *)
+ # unknown option
+ ;;
+ esac
+ shift
+done
+
+IMAGE_NAME=${IMAGE_NAME:-flink-job}
+
+# TMPDIR must be contained within the working directory so it is part of the
+# Docker context. (i.e. it can't be mktemp'd in /tmp)
+TMPDIR=_TMP_
+
+cleanup() {
+ rm -rf "${TMPDIR}"
+}
+trap cleanup EXIT
+
+mkdir -p "${TMPDIR}"
+
+JOB_JAR_TARGET="${TMPDIR}/job.jar"
+cp ${JOB_JAR_PATH} ${JOB_JAR_TARGET}
+
+if [ -n "${FROM_RELEASE}" ]; then
+
+ [[ -n "${FLINK_VERSION}" ]] && [[ -n "${HADOOP_VERSION}" ]] && [[ -n "${SCALA_VERSION}" ]] || usage
+
+ FLINK_BASE_URL="$(curl -s https://www.apache.org/dyn/closer.cgi\?preferred\=true)flink/flink-${FLINK_VERSION}/"
+ FLINK_DIST_FILE_NAME="flink-${FLINK_VERSION}-bin-hadoop${HADOOP_VERSION}-scala_${SCALA_VERSION}.tgz"
+ CURL_OUTPUT="${TMPDIR}/${FLINK_DIST_FILE_NAME}"
+
+ echo "Downloading ${FLINK_DIST_FILE_NAME} from ${FLINK_BASE_URL}"
+ curl -s ${FLINK_BASE_URL}${FLINK_DIST_FILE_NAME} --output ${CURL_OUTPUT}
--- End diff --
Good point. Let's do this improvement as a follow up.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202373877
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/EntrypointClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link EntrypointClusterConfigurationParserFactory}.
+ */
+public class EntrypointClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final int restPort = 1234;
+ final String key = "key";
+ final String value = "value";
+ final String arg1 = "arg1";
+ final String arg2 = "arg2";
+ final String[] args = {"--configDir", configDir, "-r", String.valueOf(restPort), String.format("-D%s=%s", key, value), arg1, arg2};
+
+ final EntrypointClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(restPort)));
+ final Properties dynamicProperties = clusterConfiguration.getDynamicProperties();
+
+ assertThat(dynamicProperties, hasEntry(key, value));
+
+ assertThat(clusterConfiguration.getArgs(), arrayContaining(arg1, arg2));
+ }
+
+ @Test
+ public void testOnlyRequiredArguments() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String[] args = {"--configDir", configDir};
+
+ final EntrypointClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(-1)));
+ }
+
+ @Test(expected = FlinkParseException.class)
+ public void testMissingRequiredArgument() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+ final String[] args = {};
+
+ commandLineParser.parse(args);
+ fail("Expected an FlinkParseException.");
--- End diff --
Good point. Will remove it.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202300255
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,82 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link ClusterConfigurationParserFactory}.
+ */
+public class ClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String key = "key";
+ final String value = "value";
+ final String arg1 = "arg1";
+ final String arg2 = "arg2";
+ final String[] args = {"--configDir", configDir, String.format("-D%s=%s", key, value), arg1, arg2};
+
+ final ClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ final Properties dynamicProperties = clusterConfiguration.getDynamicProperties();
+
+ assertThat(dynamicProperties, hasEntry(key, value));
+
+ assertThat(clusterConfiguration.getArgs(), arrayContaining(arg1, arg2));
+ }
+
+ @Test
+ public void testOnlyRequiredArguments() throws FlinkParseException {
+ final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String[] args = {"--configDir", configDir};
+
+ final ClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ }
+
+ @Test(expected = FlinkParseException.class)
+ public void testMissingRequiredArgument() throws FlinkParseException {
+ final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
+ final String[] args = {};
+
+ commandLineParser.parse(args);
+ fail("Expected an FlinkParseException.");
--- End diff --
Superfluous `fail`
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/6320
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202332741
--- Diff: flink-container/pom.xml ---
@@ -0,0 +1,67 @@
+<!--
--- End diff --
Licenses headers in other pom files don't use `~`.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202300519
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,82 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link ClusterConfigurationParserFactory}.
+ */
+public class ClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
--- End diff --
Could be moved to a setup method.
---
[GitHub] flink issue #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on the issue:
https://github.com/apache/flink/pull/6320
I tested the tooling around docker-compose and it works for me when using the local distribution.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202375386
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/EntrypointClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link EntrypointClusterConfigurationParserFactory}.
+ */
+public class EntrypointClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
--- End diff --
Good idea. I think we could even make it a static field because the parser is stateless.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202300463
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/EntrypointClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link EntrypointClusterConfigurationParserFactory}.
+ */
+public class EntrypointClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
--- End diff --
Could be moved to a `setUp` method.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202351378
--- Diff: flink-container/docker/README.md ---
@@ -0,0 +1,44 @@
+# Apache Flink cluster deployment on docker using docker-compose
+
+## Installation
+
+Install the most recent stable version of docker
+https://docs.docker.com/installation/
+
+## Build
+
+Images are based on the official Java Alpine (OpenJDK 8) image. If you want to
+build the flink image run:
+
+ sh build.sh --job-jar /path/to/job/jar/job.jar --image-name flink:job
+
+or
+
+ docker build -t flink .
--- End diff --
There are some args missing to make these commands work. Is it intentional?
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202305883
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfiguration.java ---
@@ -18,27 +18,43 @@
package org.apache.flink.runtime.entrypoint;
-import org.apache.flink.util.Preconditions;
+import javax.annotation.Nonnull;
+
+import java.util.Properties;
/**
* Configuration class which contains the parsed command line arguments for
* the {@link ClusterEntrypoint}.
*/
public class ClusterConfiguration {
+
+ @Nonnull
private final String configDir;
- private final int restPort;
+ @Nonnull
+ private final Properties dynamicProperties;
+
+ @Nonnull
+ private final String[] args;
- public ClusterConfiguration(String configDir, int restPort) {
- this.configDir = Preconditions.checkNotNull(configDir);
- this.restPort = restPort;
+ public ClusterConfiguration(@Nonnull String configDir, @Nonnull Properties dynamicProperties, @Nonnull String[] args) {
--- End diff --
`@Nonnull` is not used consistently, e.g., `FlinkParseException` is not annotated. I think it's easier to assume everything is non-null by default, and the rest should be annotated with `@Nullable`.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202332323
--- Diff: flink-container/src/test/java/org/apache/flink/container/entrypoint/StandaloneJobClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,89 @@
+/*
+ * 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.flink.container.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.FlinkParseException;
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link StandaloneJobClusterConfigurationParserFactory}.
+ */
+public class StandaloneJobClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<StandaloneJobClusterConfiguration> commandLineParser = new CommandLineParser<>(new StandaloneJobClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String key = "key";
+ final String value = "value";
+ final int restPort = 1234;
+ final String jobClassName = "foobar";
+ final String arg1 = "arg1";
+ final String arg2 = "arg2";
+ final String[] args = {"--configDir", configDir, "--webui-port", String.valueOf(restPort), "--job-classname", jobClassName, String.format("-D%s=%s", key, value), arg1, arg2};
+
+ final StandaloneJobClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getJobClassName(), is(equalTo(jobClassName)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(restPort)));
+ final Properties dynamicProperties = clusterConfiguration.getDynamicProperties();
+
+ assertThat(dynamicProperties, hasEntry(key, value));
+
+ assertThat(clusterConfiguration.getArgs(), arrayContaining(arg1, arg2));
+ }
+
+ @Test
+ public void testOnlyRequiredArguments() throws FlinkParseException {
+ final CommandLineParser<StandaloneJobClusterConfiguration> commandLineParser = new CommandLineParser<>(new StandaloneJobClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String jobClassName = "foobar";
+ final String[] args = {"--configDir", configDir, "--job-classname", jobClassName};
+
+ final StandaloneJobClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getJobClassName(), is(equalTo(jobClassName)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(-1)));
+ }
+
+ @Test(expected = FlinkParseException.class)
+ public void testMissingRequiredArgument() throws FlinkParseException {
+ final CommandLineParser<StandaloneJobClusterConfiguration> commandLineParser = new CommandLineParser<>(new StandaloneJobClusterConfigurationParserFactory());
+ final String[] args = {};
+
+ commandLineParser.parse(args);
+ fail("Expected an FlinkParseException.");
--- End diff --
Superfluous `fail`
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202376168
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfiguration.java ---
@@ -18,27 +18,43 @@
package org.apache.flink.runtime.entrypoint;
-import org.apache.flink.util.Preconditions;
+import javax.annotation.Nonnull;
+
+import java.util.Properties;
/**
* Configuration class which contains the parsed command line arguments for
* the {@link ClusterEntrypoint}.
*/
public class ClusterConfiguration {
+
+ @Nonnull
private final String configDir;
- private final int restPort;
+ @Nonnull
+ private final Properties dynamicProperties;
+
+ @Nonnull
+ private final String[] args;
- public ClusterConfiguration(String configDir, int restPort) {
- this.configDir = Preconditions.checkNotNull(configDir);
- this.restPort = restPort;
+ public ClusterConfiguration(@Nonnull String configDir, @Nonnull Properties dynamicProperties, @Nonnull String[] args) {
--- End diff --
In `FlinkParseException` it does not follow the convention because the parent class also does not follow the convention. In the `ClusterConfiguration` the usage of `@Nonnull` should be consistent and allows us to not explicitly have to check for non null arguments.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202376612
--- Diff: flink-container/docker/README.md ---
@@ -0,0 +1,44 @@
+# Apache Flink cluster deployment on docker using docker-compose
+
+## Installation
+
+Install the most recent stable version of docker
+https://docs.docker.com/installation/
+
+## Build
+
+Images are based on the official Java Alpine (OpenJDK 8) image. If you want to
+build the flink image run:
+
+ sh build.sh --job-jar /path/to/job/jar/job.jar --image-name flink:job
--- End diff --
True. I will remove it.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202336290
--- Diff: flink-container/docker/README.md ---
@@ -0,0 +1,44 @@
+# Apache Flink cluster deployment on docker using docker-compose
+
+## Installation
+
+Install the most recent stable version of docker
+https://docs.docker.com/installation/
+
+## Build
+
+Images are based on the official Java Alpine (OpenJDK 8) image. If you want to
+build the flink image run:
+
+ sh build.sh --job-jar /path/to/job/jar/job.jar --image-name flink:job
--- End diff --
nit: The shebang in `build.sh` already tells the program loader to use `sh`.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202299810
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/EntrypointClusterConfigurationParserFactoryTest.java ---
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.runtime.entrypoint;
+
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link EntrypointClusterConfigurationParserFactory}.
+ */
+public class EntrypointClusterConfigurationParserFactoryTest extends TestLogger {
+
+ @Test
+ public void testEntrypointClusterConfigurationParsing() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final int restPort = 1234;
+ final String key = "key";
+ final String value = "value";
+ final String arg1 = "arg1";
+ final String arg2 = "arg2";
+ final String[] args = {"--configDir", configDir, "-r", String.valueOf(restPort), String.format("-D%s=%s", key, value), arg1, arg2};
+
+ final EntrypointClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(restPort)));
+ final Properties dynamicProperties = clusterConfiguration.getDynamicProperties();
+
+ assertThat(dynamicProperties, hasEntry(key, value));
+
+ assertThat(clusterConfiguration.getArgs(), arrayContaining(arg1, arg2));
+ }
+
+ @Test
+ public void testOnlyRequiredArguments() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+
+ final String configDir = "/foo/bar";
+ final String[] args = {"--configDir", configDir};
+
+ final EntrypointClusterConfiguration clusterConfiguration = commandLineParser.parse(args);
+
+ assertThat(clusterConfiguration.getConfigDir(), is(equalTo(configDir)));
+ assertThat(clusterConfiguration.getRestPort(), is(equalTo(-1)));
+ }
+
+ @Test(expected = FlinkParseException.class)
+ public void testMissingRequiredArgument() throws FlinkParseException {
+ final CommandLineParser<EntrypointClusterConfiguration> commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());
+ final String[] args = {};
+
+ commandLineParser.parse(args);
+ fail("Expected an FlinkParseException.");
--- End diff --
With `fail` is superfluous here.
```
@Test(expected = FlinkParseException.class)
public void testMissingRequiredArgument() throws FlinkParseException {
try {
final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
final String[] args = {};
commandLineParser.parse(args);
} catch (Exception e) {}
}
```
would already fail with:
```
java.lang.AssertionError: Expected exception: org.apache.flink.runtime.entrypoint.FlinkParseException
at org.junit.internal.runners.statements.ExpectException.evaluate(ExpectException.java:32)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
```
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202376970
--- Diff: flink-container/docker/README.md ---
@@ -0,0 +1,44 @@
+# Apache Flink cluster deployment on docker using docker-compose
+
+## Installation
+
+Install the most recent stable version of docker
+https://docs.docker.com/installation/
+
+## Build
+
+Images are based on the official Java Alpine (OpenJDK 8) image. If you want to
+build the flink image run:
+
+ sh build.sh --job-jar /path/to/job/jar/job.jar --image-name flink:job
+
+or
+
+ docker build -t flink .
--- End diff --
No not really. I will update the documentation. Thanks for catching it.
---
[GitHub] flink issue #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/6320
Thanks for the review @GJL. I've addressed your comments. Merging this PR.
---
[GitHub] flink pull request #6320: [FLINK-9823] Add Kubernetes deployment ymls
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/6320#discussion_r202348473
--- Diff: flink-container/docker/build.sh ---
@@ -0,0 +1,128 @@
+#!/bin/sh
+
+################################################################################
+# 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.
+################################################################################
+
+usage() {
+ cat <<HERE
+Usage:
+ build.sh --job-jar <path-to-job-jar> --from-local-dist [--image-name <image>]
+ build.sh --job-jar <path-to-job-jar> --from-archive <path-to-dist-archive> [--image-name <image>]
+ build.sh --job-jar <path-to-job-jar> --from-release --flink-version <x.x.x> --hadoop-version <x.x> --scala-version <x.xx> [--image-name <image>]
+ build.sh --help
+
+ If the --image-name flag is not used the built image name will be 'flink'.
+HERE
+ exit 1
+}
+
+while [[ $# -ge 1 ]]
+do
+key="$1"
+ case $key in
+ --job-jar)
+ JOB_JAR_PATH="$2"
+ shift
+ ;;
+ --from-local-dist)
+ FROM_LOCAL="true"
+ ;;
+ --from-archive)
+ FROM_ARCHIVE="$2"
+ shift
+ ;;
+ --from-release)
+ FROM_RELEASE="true"
+ ;;
+ --image-name)
+ IMAGE_NAME="$2"
+ shift
+ ;;
+ --flink-version)
+ FLINK_VERSION="$2"
+ shift
+ ;;
+ --hadoop-version)
+ HADOOP_VERSION="$(echo "$2" | sed 's/\.//')"
+ shift
+ ;;
+ --scala-version)
+ SCALA_VERSION="$2"
+ shift
+ ;;
+ --kubernetes-certificates)
+ CERTIFICATES_DIR="$2"
+ shift
+ ;;
+ --help)
+ usage
+ ;;
+ *)
+ # unknown option
+ ;;
+ esac
+ shift
+done
+
+IMAGE_NAME=${IMAGE_NAME:-flink-job}
+
+# TMPDIR must be contained within the working directory so it is part of the
+# Docker context. (i.e. it can't be mktemp'd in /tmp)
+TMPDIR=_TMP_
+
+cleanup() {
+ rm -rf "${TMPDIR}"
+}
+trap cleanup EXIT
+
+mkdir -p "${TMPDIR}"
+
+JOB_JAR_TARGET="${TMPDIR}/job.jar"
+cp ${JOB_JAR_PATH} ${JOB_JAR_TARGET}
+
+if [ -n "${FROM_RELEASE}" ]; then
+
+ [[ -n "${FLINK_VERSION}" ]] && [[ -n "${HADOOP_VERSION}" ]] && [[ -n "${SCALA_VERSION}" ]] || usage
+
+ FLINK_BASE_URL="$(curl -s https://www.apache.org/dyn/closer.cgi\?preferred\=true)flink/flink-${FLINK_VERSION}/"
+ FLINK_DIST_FILE_NAME="flink-${FLINK_VERSION}-bin-hadoop${HADOOP_VERSION}-scala_${SCALA_VERSION}.tgz"
+ CURL_OUTPUT="${TMPDIR}/${FLINK_DIST_FILE_NAME}"
+
+ echo "Downloading ${FLINK_DIST_FILE_NAME} from ${FLINK_BASE_URL}"
+ curl -s ${FLINK_BASE_URL}${FLINK_DIST_FILE_NAME} --output ${CURL_OUTPUT}
--- End diff --
I think it's a bit surprising that for 1.5.0 the command will fail:
`build.sh --job-jar ./job.jar --from-release --flink-version 1.5.0 --hadoop-version 28 --scala-version 2.11`
`build.sh --job-jar ./job.jar --from-release --flink-version 1.5.1 --hadoop-version 28 --scala-version 2.11`
---