You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/06/13 10:08:53 UTC
[flink] branch master updated (00d90a4 -> 9e6ff81)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 00d90a4 [FLINK-12788][container] Add support for Python jobs in build script
new d14117f [hotfix][dist] Rename FLINK_ROOT*_DIR to FLINK_HOME
new 9871f25 [hotfix][dist] Export variables in only one place (config.sh)
new 6ed28c8 [hotfix][runtime] Deduplicate and simplify environment variables handling in FlinkDistributionOverlay
new a98a7ee [hotfix][core] Deduplicate code and unify method names in GlobalConfiguration
new b5dc213 [FLINK-12143] Distribute FLINK_PLUGINS_DIR the same way as FLINK_LIB_DIR is distributed
new 186a2dd [FLINK-12143][dist] Ship plugins in the cluster the same way as lib jars
new 7020f62 [hotfix][dist] Use proper flink-python jar match pattern in pyflink.bat for Windows (FLINK-12409)
new 33b182c [FLINK-12143] PluginLoader: use system level always parents first patterns in class loading
new 0075361 [FLINK-12143][e2e] Introduce plugins helper function and test cleanup
new 9e6ff81 [FLINK-12143][e2e] Change end-to-end tests with file system to use plugins mechanism
The 10 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:
.../org/apache/flink/client/cli/CliFrontend.java | 4 +-
flink-container/docker/Dockerfile | 1 +
.../flink/configuration/ConfigConstants.java | 3 +
.../apache/flink/configuration/CoreOptions.java | 8 +--
.../flink/configuration/GlobalConfiguration.java | 49 +++++++++++-----
.../org/apache/flink/core/plugin/PluginConfig.java | 68 ++++++++++++++++++++++
.../org/apache/flink/core/plugin/PluginLoader.java | 5 +-
.../apache/flink/core/plugin/PluginManager.java | 14 +++--
.../org/apache/flink/core/plugin/PluginUtils.java | 23 ++++----
.../{IterableIterator.java => ArrayUtils.java} | 21 +++++--
.../configuration/GlobalConfigurationTest.java | 2 +-
.../java/org/apache/flink/util/ArrayUtilsTest.java | 53 +++++++++++++++++
flink-dist/src/main/flink-bin/bin/config.sh | 30 +++++-----
flink-dist/src/main/flink-bin/bin/flink | 3 -
flink-dist/src/main/flink-bin/bin/flink.bat | 5 +-
.../main/flink-bin/bin/pyflink-gateway-server.sh | 4 +-
.../src/main/flink-bin/bin/pyflink-stream.sh | 2 +-
flink-dist/src/main/flink-bin/bin/pyflink.bat | 4 +-
flink-dist/src/main/flink-bin/bin/pyflink.sh | 2 +-
.../src/main/flink-bin/bin/start-cluster.bat | 9 +--
.../flink-bin/mesos-bin/mesos-appmaster-job.sh | 4 --
.../main/flink-bin/mesos-bin/mesos-appmaster.sh | 4 --
.../main/flink-bin/mesos-bin/mesos-taskmanager.sh | 4 --
.../src/main/flink-bin/yarn-bin/yarn-session.sh | 2 -
flink-end-to-end-tests/test-scripts/common.sh | 15 +++++
...test_sql_client_kafka.sh => common_dummy_fs.sh} | 7 ++-
flink-end-to-end-tests/test-scripts/common_s3.sh | 2 +-
.../test-scripts/test_azure_fs.sh | 2 +-
.../test-scripts/test_batch_wordcount.sh | 3 +-
.../test-scripts/test_docker_embedded_job.sh | 3 +-
.../test-scripts/test_yarn_kerberos_docker.sh | 3 +-
.../mesos/entrypoint/MesosEntrypointUtils.java | 2 +-
.../mesos/entrypoint/MesosTaskExecutorRunner.java | 4 +-
.../runtime/webmonitor/history/HistoryServer.java | 4 +-
.../overlays/FlinkDistributionOverlay.java | 44 ++++++--------
.../runtime/entrypoint/ClusterEntrypoint.java | 4 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 4 +-
.../overlays/FlinkDistributionOverlayTest.java | 25 ++++++--
flink-table/flink-sql-client/bin/sql-client.sh | 3 -
.../table/client/gateway/local/LocalExecutor.java | 4 +-
.../apache/flink/test/plugin/PluginLoaderTest.java | 6 +-
.../flink/test/plugin/PluginManagerTest.java | 2 +-
.../flink/yarn/AbstractYarnClusterDescriptor.java | 34 +++++++----
.../apache/flink/yarn/YarnTaskExecutorRunner.java | 4 +-
.../flink/yarn/YarnClusterDescriptorTest.java | 22 ++++---
45 files changed, 347 insertions(+), 174 deletions(-)
create mode 100644 flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
copy flink-core/src/main/java/org/apache/flink/util/{IterableIterator.java => ArrayUtils.java} (66%)
create mode 100644 flink-core/src/test/java/org/apache/flink/util/ArrayUtilsTest.java
copy flink-end-to-end-tests/test-scripts/{test_sql_client_kafka.sh => common_dummy_fs.sh} (84%)
mode change 100755 => 100644
[flink] 06/10: [FLINK-12143][dist] Ship plugins in the cluster the
same way as lib jars
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 186a2dd6559653fc9573b9cdb5221347417b65a8
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Apr 26 15:46:12 2019 +0200
[FLINK-12143][dist] Ship plugins in the cluster the same way as lib jars
---
.../org/apache/flink/client/cli/CliFrontend.java | 4 +-
.../flink/configuration/GlobalConfiguration.java | 25 ++++++++++++
.../plugin/{PluginUtils.java => PluginConfig.java} | 46 ++++++++++++----------
.../org/apache/flink/core/plugin/PluginUtils.java | 23 +++++------
.../configuration/GlobalConfigurationTest.java | 2 +-
.../mesos/entrypoint/MesosTaskExecutorRunner.java | 4 +-
.../runtime/webmonitor/history/HistoryServer.java | 4 +-
.../runtime/entrypoint/ClusterEntrypoint.java | 4 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 4 +-
.../table/client/gateway/local/LocalExecutor.java | 4 +-
.../flink/yarn/AbstractYarnClusterDescriptor.java | 6 +--
.../apache/flink/yarn/YarnTaskExecutorRunner.java | 4 +-
12 files changed, 73 insertions(+), 57 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index 06dd761..ec67fff 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -80,7 +80,6 @@ import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -124,8 +123,7 @@ public class CliFrontend {
this.configuration = Preconditions.checkNotNull(configuration);
this.customCommandLines = Preconditions.checkNotNull(customCommandLines);
- //TODO provide plugin path.
- FileSystem.initialize(this.configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(configuration));
this.customCommandLineOptions = new Options();
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java b/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
index 564fe19..c0debab 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
@@ -130,9 +130,34 @@ public final class GlobalConfiguration {
configuration.addAll(dynamicProperties);
}
+ return enrichWithEnvironmentVariables(configuration);
+ }
+
+ private static Configuration enrichWithEnvironmentVariables(Configuration configuration) {
+ enrichWithEnvironmentVariable(ConfigConstants.ENV_FLINK_PLUGINS_DIR, configuration);
return configuration;
}
+ private static void enrichWithEnvironmentVariable(String environmentVariable, Configuration configuration) {
+ String pluginsDirFromEnv = System.getenv(environmentVariable);
+
+ if (pluginsDirFromEnv == null) {
+ return;
+ }
+
+ String pluginsDirFromConfig = configuration.getString(environmentVariable, pluginsDirFromEnv);
+
+ if (!pluginsDirFromEnv.equals(pluginsDirFromConfig)) {
+ throw new IllegalConfigurationException(
+ "The given configuration file already contains a value (" + pluginsDirFromEnv +
+ ") for the key (" + environmentVariable +
+ ") that would have been overwritten with (" + pluginsDirFromConfig +
+ ") by an environment with the same name.");
+ }
+
+ configuration.setString(environmentVariable, pluginsDirFromEnv);
+ }
+
/**
* Loads a YAML-file of key-value pairs.
*
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
similarity index 51%
copy from flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
copy to flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
index 17e2ae4..c4a3b2b 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
@@ -18,37 +18,41 @@
package org.apache.flink.core.plugin;
-import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
-import java.io.IOException;
+import java.io.File;
import java.nio.file.Path;
-import java.util.Collection;
-import java.util.Collections;
import java.util.Optional;
/**
- * Utility functions for the plugin mechanism.
+ * Stores the configuration for plugins mechanism.
*/
-public final class PluginUtils {
+public class PluginConfig {
+ private final Optional<Path> pluginsPath;
- private PluginUtils() {
- throw new AssertionError("Singleton class.");
+ private PluginConfig() {
+ this.pluginsPath = Optional.empty();
}
- public static PluginManager createPluginManagerFromRootFolder(Optional<Path> pluginsRootPath) {
- Collection<PluginDescriptor> pluginDescriptorsForDirectory;
-
- if (pluginsRootPath.isPresent()) {
- try {
- pluginDescriptorsForDirectory =
- new DirectoryBasedPluginFinder(pluginsRootPath.get()).findPlugins();
- } catch (IOException e) {
- throw new FlinkRuntimeException("Exception when trying to initialize plugin system.", e);
- }
- } else {
- pluginDescriptorsForDirectory = Collections.emptyList();
+ private PluginConfig(Path pluginsPath) {
+ this.pluginsPath = Optional.of(pluginsPath);
+ }
+
+ public Optional<Path> getPluginsPath() {
+ return pluginsPath;
+ }
+
+ public static PluginConfig fromConfiguration(Configuration configuration) {
+ String pluginsDir = configuration.getString(ConfigConstants.ENV_FLINK_PLUGINS_DIR, null);
+ if (pluginsDir == null) {
+ return new PluginConfig();
}
- return new PluginManager(pluginDescriptorsForDirectory);
+ File pluginsDirFile = new File(pluginsDir);
+ if (!pluginsDirFile.isDirectory()) {
+ return new PluginConfig();
+ }
+ return new PluginConfig(pluginsDirFile.toPath());
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
index 17e2ae4..0f9f0d3 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
@@ -18,13 +18,12 @@
package org.apache.flink.core.plugin;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.FlinkRuntimeException;
import java.io.IOException;
-import java.nio.file.Path;
import java.util.Collection;
import java.util.Collections;
-import java.util.Optional;
/**
* Utility functions for the plugin mechanism.
@@ -35,20 +34,22 @@ public final class PluginUtils {
throw new AssertionError("Singleton class.");
}
- public static PluginManager createPluginManagerFromRootFolder(Optional<Path> pluginsRootPath) {
- Collection<PluginDescriptor> pluginDescriptorsForDirectory;
+ public static PluginManager createPluginManagerFromRootFolder(Configuration configuration) {
+ return createPluginManagerFromRootFolder(PluginConfig.fromConfiguration(configuration));
+ }
- if (pluginsRootPath.isPresent()) {
+ private static PluginManager createPluginManagerFromRootFolder(PluginConfig pluginConfig) {
+ if (pluginConfig.getPluginsPath().isPresent()) {
try {
- pluginDescriptorsForDirectory =
- new DirectoryBasedPluginFinder(pluginsRootPath.get()).findPlugins();
+ Collection<PluginDescriptor> pluginDescriptors =
+ new DirectoryBasedPluginFinder(pluginConfig.getPluginsPath().get()).findPlugins();
+ return new PluginManager(pluginDescriptors);
} catch (IOException e) {
throw new FlinkRuntimeException("Exception when trying to initialize plugin system.", e);
}
- } else {
- pluginDescriptorsForDirectory = Collections.emptyList();
}
-
- return new PluginManager(pluginDescriptorsForDirectory);
+ else {
+ return new PluginManager(Collections.emptyList());
+ }
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/GlobalConfigurationTest.java b/flink-core/src/test/java/org/apache/flink/configuration/GlobalConfigurationTest.java
index a42564d..94bb872 100644
--- a/flink-core/src/test/java/org/apache/flink/configuration/GlobalConfigurationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/configuration/GlobalConfigurationTest.java
@@ -97,7 +97,7 @@ public class GlobalConfigurationTest extends TestLogger {
@Test(expected = IllegalArgumentException.class)
public void testFailIfNull() {
- GlobalConfiguration.loadConfiguration(null);
+ GlobalConfiguration.loadConfiguration((String) null);
}
@Test(expected = IllegalConfigurationException.class)
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
index e2d0161..4c0ec97 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
@@ -43,7 +43,6 @@ import org.slf4j.LoggerFactory;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.Map;
-import java.util.Optional;
/**
* The entry point for running a TaskManager in a Mesos container.
@@ -87,8 +86,7 @@ public class MesosTaskExecutorRunner {
final Map<String, String> envs = System.getenv();
// configure the filesystems
- //TODO provide plugin path.
- FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(configuration));
// tell akka to die in case of an error
configuration.setBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR, true);
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
index 12a5b56..e907836 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
@@ -58,7 +58,6 @@ import java.nio.file.Files;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.List;
-import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
@@ -111,8 +110,7 @@ public class HistoryServer {
LOG.info("Loading configuration from {}", configDir);
final Configuration flinkConfig = GlobalConfiguration.loadConfiguration(configDir);
- //TODO provide plugin path.
- FileSystem.initialize(flinkConfig, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(flinkConfig, PluginUtils.createPluginManagerFromRootFolder(flinkConfig));
// run the history server
SecurityUtils.install(new SecurityConfiguration(flinkConfig));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index d6dc724..0538bec 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -77,7 +77,6 @@ import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
@@ -187,8 +186,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro
private void configureFileSystems(Configuration configuration) {
LOG.info("Install default filesystem.");
- //TODO provide plugin path
- FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(configuration));
}
protected SecurityContext installSecurityContext(Configuration configuration) throws Exception {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index d5985b1..2b4e0f2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -70,7 +70,6 @@ import java.lang.reflect.UndeclaredThrowableException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -285,8 +284,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
final Configuration configuration = loadConfiguration(args);
- //TODO provide plugin path.
- FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(configuration));
SecurityUtils.install(new SecurityConfiguration(configuration));
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index 158c261..99939bd 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -69,7 +69,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
/**
* Executor that performs the Flink communication locally. The calls are blocking depending on the
@@ -113,8 +112,7 @@ public class LocalExecutor implements Executor {
this.flinkConfig = GlobalConfiguration.loadConfiguration(flinkConfigDir);
// initialize default file system
- //TODO provide plugin path.
- FileSystem.initialize(flinkConfig, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(flinkConfig, PluginUtils.createPluginManagerFromRootFolder(flinkConfig));
// load command lines for deployment
this.commandLines = CliFrontend.loadCustomCommandLines(flinkConfig, flinkConfigDir);
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index ebf470c..8b51968 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -97,7 +97,6 @@ import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
@@ -660,8 +659,9 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
// ------------------ Initialize the file systems -------------------------
- //TODO provide plugin path.
- org.apache.flink.core.fs.FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ org.apache.flink.core.fs.FileSystem.initialize(
+ configuration,
+ PluginUtils.createPluginManagerFromRootFolder(configuration));
// initialize file system
// Copy the application master jar to the filesystem
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
index d1a3700..bcb87e1 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
@@ -46,7 +46,6 @@ import java.io.File;
import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.Map;
-import java.util.Optional;
import java.util.concurrent.Callable;
/**
@@ -95,8 +94,7 @@ public class YarnTaskExecutorRunner {
final Configuration configuration = GlobalConfiguration.loadConfiguration(currDir);
- //TODO provide path.
- FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(Optional.empty()));
+ FileSystem.initialize(configuration, PluginUtils.createPluginManagerFromRootFolder(configuration));
setupConfigurationAndInstallSecurityContext(configuration, currDir, ENV);
[flink] 04/10: [hotfix][core] Deduplicate code and unify method
names in GlobalConfiguration
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a98a7ee8ba716844372ebf4dc2b8efb3dcbd6cb3
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Apr 26 15:48:51 2019 +0200
[hotfix][core] Deduplicate code and unify method names in GlobalConfiguration
---
.../flink/configuration/GlobalConfiguration.java | 32 ++++++++++------------
.../mesos/entrypoint/MesosEntrypointUtils.java | 2 +-
2 files changed, 15 insertions(+), 19 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java b/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
index fd7d441..564fe19 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java
@@ -63,11 +63,23 @@ public final class GlobalConfiguration {
* @return Returns the Configuration
*/
public static Configuration loadConfiguration() {
+ return loadConfiguration(new Configuration());
+ }
+
+ /**
+ * Loads the global configuration and adds the given dynamic properties
+ * configuration.
+ *
+ * @param dynamicProperties The given dynamic properties
+ * @return Returns the loaded global configuration with dynamic properties
+ */
+ public static Configuration loadConfiguration(Configuration dynamicProperties) {
final String configDir = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
if (configDir == null) {
- return new Configuration();
+ return new Configuration(dynamicProperties);
}
- return loadConfiguration(configDir, null);
+
+ return loadConfiguration(configDir, dynamicProperties);
}
/**
@@ -122,22 +134,6 @@ public final class GlobalConfiguration {
}
/**
- * Loads the global configuration and adds the given dynamic properties
- * configuration.
- *
- * @param dynamicProperties The given dynamic properties
- * @return Returns the loaded global configuration with dynamic properties
- */
- public static Configuration loadConfigurationWithDynamicProperties(Configuration dynamicProperties) {
- final String configDir = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
- if (configDir == null) {
- return new Configuration(dynamicProperties);
- }
-
- return loadConfiguration(configDir, dynamicProperties);
- }
-
- /**
* Loads a YAML-file of key-value pairs.
*
* <p>Colon and whitespace ": " separate key and value (one per line). The hash tag "#" starts a single-line comment.
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosEntrypointUtils.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosEntrypointUtils.java
index 4d7a485..19979e4 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosEntrypointUtils.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosEntrypointUtils.java
@@ -168,7 +168,7 @@ public class MesosEntrypointUtils {
*/
public static Configuration loadConfiguration(Configuration dynamicProperties, Logger log) {
Configuration configuration =
- GlobalConfiguration.loadConfigurationWithDynamicProperties(dynamicProperties);
+ GlobalConfiguration.loadConfiguration(dynamicProperties);
// read the environment variables
final Map<String, String> envs = System.getenv();
[flink] 09/10: [FLINK-12143][e2e] Introduce plugins helper function
and test cleanup
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 007536169be09b0ae7cd0d2921f507958fff78cb
Author: Aleksey Pak <al...@ververica.com>
AuthorDate: Thu Jun 6 12:09:42 2019 +0200
[FLINK-12143][e2e] Introduce plugins helper function and test cleanup
---
flink-end-to-end-tests/test-scripts/common.sh | 15 +++++++++++++++
1 file changed, 15 insertions(+)
diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 1a277e7..361e49a 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -99,6 +99,9 @@ function revert_flink_dir() {
rm -r "${TEST_DATA_DIR}/tmp/backup"
+ # By default, the plugins dir doesn't exist. Some tests may have created it.
+ rm -r "${FLINK_DIR}/plugins"
+
REST_PROTOCOL="http"
CURL_SSL_ARGS=""
}
@@ -108,6 +111,18 @@ function add_optional_lib() {
cp "$FLINK_DIR/opt/flink-${lib_name}"*".jar" "$FLINK_DIR/lib"
}
+function add_optional_plugin() {
+ # This is similar to add_optional_lib, but the jar would be copied to
+ # Flink's plugins dir (the nested folder name does not matter).
+ # Note: this may not work with some jars, as not all of them implement plugin api.
+ # Please check the corresponding code of the jar.
+ local plugin="$1"
+ local plugin_dir="$FLINK_DIR/plugins/$plugin"
+
+ mkdir -p "$plugin_dir"
+ cp "$FLINK_DIR/opt/flink-$plugin"*".jar" "$plugin_dir"
+}
+
function delete_config_key() {
local config_key=$1
sed -i -e "/^${config_key}: /d" ${FLINK_DIR}/conf/flink-conf.yaml
[flink] 07/10: [hotfix][dist] Use proper flink-python jar match
pattern in pyflink.bat for Windows (FLINK-12409)
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7020f62439200a5be2ff6b308955ee44160559fd
Author: Aleksey Pak <al...@ververica.com>
AuthorDate: Wed Jun 5 17:40:15 2019 +0200
[hotfix][dist] Use proper flink-python jar match pattern in pyflink.bat for Windows (FLINK-12409)
---
flink-dist/src/main/flink-bin/bin/pyflink.bat | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink.bat b/flink-dist/src/main/flink-bin/bin/pyflink.bat
index 65c4f21..de383e9 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink.bat
+++ b/flink-dist/src/main/flink-bin/bin/pyflink.bat
@@ -22,4 +22,4 @@ setlocal EnableDelayedExpansion
SET bin=%~dp0
SET FLINK_HOME=%bin%..
-"%FLINK_HOME%\bin\flink" run -v "%FLINK_HOME%"\opt\flink-python*.jar %*
+"%FLINK_HOME%\bin\flink" run -v "%FLINK_HOME%"\opt\flink-python_*.jar %*
[flink] 05/10: [FLINK-12143] Distribute FLINK_PLUGINS_DIR the same
way as FLINK_LIB_DIR is distributed
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b5dc213ea7a0bdbf021a10cd54ec452003bb5173
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Apr 24 18:23:56 2019 +0200
[FLINK-12143] Distribute FLINK_PLUGINS_DIR the same way as FLINK_LIB_DIR is distributed
---
flink-container/docker/Dockerfile | 1 +
.../flink/configuration/ConfigConstants.java | 3 +++
flink-dist/src/main/flink-bin/bin/config.sh | 2 ++
flink-dist/src/main/flink-bin/bin/flink.bat | 1 +
.../src/main/flink-bin/bin/start-cluster.bat | 1 +
.../overlays/FlinkDistributionOverlay.java | 12 +++++++---
.../overlays/FlinkDistributionOverlayTest.java | 25 ++++++++++++++-----
.../flink/yarn/AbstractYarnClusterDescriptor.java | 28 +++++++++++++++-------
.../flink/yarn/YarnClusterDescriptorTest.java | 22 ++++++++++-------
9 files changed, 69 insertions(+), 26 deletions(-)
diff --git a/flink-container/docker/Dockerfile b/flink-container/docker/Dockerfile
index 853ac36..129b5c5 100644
--- a/flink-container/docker/Dockerfile
+++ b/flink-container/docker/Dockerfile
@@ -25,6 +25,7 @@ RUN apk add --no-cache bash snappy libc6-compat
ENV FLINK_INSTALL_PATH=/opt
ENV FLINK_HOME $FLINK_INSTALL_PATH/flink
ENV FLINK_LIB_DIR $FLINK_HOME/lib
+ENV FLINK_PLUGINS_DIR $FLINK_HOME/plugins
ENV FLINK_OPT_DIR $FLINK_HOME/opt
ENV FLINK_JOB_ARTIFACTS_DIR $FLINK_INSTALL_PATH/artifacts
ENV PATH $PATH:$FLINK_HOME/bin
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index a4de6cc..082ec56 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -2011,6 +2011,9 @@ public final class ConfigConstants {
/** The environment variable name which contains the location of the opt directory. */
public static final String ENV_FLINK_OPT_DIR = "FLINK_OPT_DIR";
+ /** The environment variable name which contains the location of the plugins folder. */
+ public static final String ENV_FLINK_PLUGINS_DIR = "FLINK_PLUGINS_DIR";
+
/** The environment variable name which contains the location of the bin directory. */
public static final String ENV_FLINK_BIN_DIR = "FLINK_BIN_DIR";
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index e430e91..79fa6ad 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -298,6 +298,7 @@ SYMLINK_RESOLVED_BIN=`cd "$bin"; pwd -P`
# Define the main directory of the flink installation
FLINK_HOME=`dirname "$SYMLINK_RESOLVED_BIN"`
FLINK_LIB_DIR=$FLINK_HOME/lib
+FLINK_PLUGINS_DIR=$FLINK_HOME/plugins
FLINK_OPT_DIR=$FLINK_HOME/opt
@@ -314,6 +315,7 @@ YAML_CONF=${FLINK_CONF_DIR}/${FLINK_CONF_FILE}
### Exported environment variables ###
export FLINK_CONF_DIR
export FLINK_BIN_DIR
+export FLINK_PLUGINS_DIR
# export /lib dir to access it during deployment of the Yarn staging files
export FLINK_LIB_DIR
# export /opt dir to access it for the SQL client
diff --git a/flink-dist/src/main/flink-bin/bin/flink.bat b/flink-dist/src/main/flink-bin/bin/flink.bat
index 8ed1fcf..a88a4db 100644
--- a/flink-dist/src/main/flink-bin/bin/flink.bat
+++ b/flink-dist/src/main/flink-bin/bin/flink.bat
@@ -22,6 +22,7 @@ setlocal
SET bin=%~dp0
SET FLINK_HOME=%bin%..
SET FLINK_LIB_DIR=%FLINK_HOME%\lib
+SET FLINK_PLUGINS_DIR=%FLINK_HOME%\plugins
SET JVM_ARGS=-Xmx512m
diff --git a/flink-dist/src/main/flink-bin/bin/start-cluster.bat b/flink-dist/src/main/flink-bin/bin/start-cluster.bat
index a852a55..9c01cb2 100644
--- a/flink-dist/src/main/flink-bin/bin/start-cluster.bat
+++ b/flink-dist/src/main/flink-bin/bin/start-cluster.bat
@@ -22,6 +22,7 @@ setlocal EnableDelayedExpansion
SET bin=%~dp0
SET FLINK_HOME=%bin%..
SET FLINK_LIB_DIR=%FLINK_HOME%\lib
+SET FLINK_PLUGINS_DIR=%FLINK_HOME%\plugins
SET FLINK_CONF_DIR=%FLINK_HOME%\conf
SET FLINK_LOG_DIR=%FLINK_HOME%\log
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
index baebd8c..a430e0c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
@@ -21,17 +21,18 @@ package org.apache.flink.runtime.clusterframework.overlays;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.util.Map;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_BIN_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_HOME_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_LIB_DIR;
+import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_PLUGINS_DIR;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -56,11 +57,13 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
final File flinkBinPath;
final File flinkConfPath;
final File flinkLibPath;
+ final File flinkPluginsPath;
- public FlinkDistributionOverlay(File flinkBinPath, File flinkConfPath, File flinkLibPath) {
+ public FlinkDistributionOverlay(File flinkBinPath, File flinkConfPath, File flinkLibPath, File flinkPluginsPath) {
this.flinkBinPath = checkNotNull(flinkBinPath);
this.flinkConfPath = checkNotNull(flinkConfPath);
this.flinkLibPath = checkNotNull(flinkLibPath);
+ this.flinkPluginsPath = checkNotNull(flinkPluginsPath);
}
@Override
@@ -72,6 +75,7 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
addPathRecursively(flinkBinPath, TARGET_ROOT, container);
addPathRecursively(flinkConfPath, TARGET_ROOT, container);
addPathRecursively(flinkLibPath, TARGET_ROOT, container);
+ addPathRecursively(flinkPluginsPath, TARGET_ROOT, container);
}
public static Builder newBuilder() {
@@ -85,6 +89,7 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
File flinkBinPath;
File flinkConfPath;
File flinkLibPath;
+ File flinkPluginsPath;
/**
* Configures the overlay using the current environment.
@@ -97,12 +102,13 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
flinkBinPath = getObligatoryFileFromEnvironment(ENV_FLINK_BIN_DIR);
flinkConfPath = getObligatoryFileFromEnvironment(ENV_FLINK_CONF_DIR);
flinkLibPath = getObligatoryFileFromEnvironment(ENV_FLINK_LIB_DIR);
+ flinkPluginsPath = getObligatoryFileFromEnvironment(ENV_FLINK_PLUGINS_DIR);
return this;
}
public FlinkDistributionOverlay build() {
- return new FlinkDistributionOverlay(flinkBinPath, flinkConfPath, flinkLibPath);
+ return new FlinkDistributionOverlay(flinkBinPath, flinkConfPath, flinkLibPath, flinkPluginsPath);
}
private static File getObligatoryFileFromEnvironment(String envVariableName) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlayTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlayTest.java
index e77dd3a..ef7a5e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlayTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlayTest.java
@@ -30,6 +30,7 @@ import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_PLUGINS_DIR;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@@ -49,6 +50,7 @@ public class FlinkDistributionOverlayTest extends ContainerOverlayTestBase {
File binFolder = tempFolder.newFolder("bin");
File libFolder = tempFolder.newFolder("lib");
+ File pluginsFolder = tempFolder.newFolder("plugins");
File confFolder = tempFolder.newFolder("conf");
Path[] files = createPaths(
@@ -58,14 +60,17 @@ public class FlinkDistributionOverlayTest extends ContainerOverlayTestBase {
"lib/foo.jar",
"lib/A/foo.jar",
"lib/B/foo.jar",
- "lib/B/bar.jar");
+ "lib/B/bar.jar",
+ "plugins/P1/plugin1a.jar",
+ "plugins/P1/plugin1b.jar",
+ "plugins/P2/plugin2.jar");
ContainerSpecification containerSpecification = new ContainerSpecification();
FlinkDistributionOverlay overlay = new FlinkDistributionOverlay(
binFolder,
confFolder,
- libFolder
- );
+ libFolder,
+ pluginsFolder);
overlay.configure(containerSpecification);
for(Path file : files) {
@@ -79,12 +84,14 @@ public class FlinkDistributionOverlayTest extends ContainerOverlayTestBase {
File binFolder = tempFolder.newFolder("bin");
File libFolder = tempFolder.newFolder("lib");
+ File pluginsFolder = tempFolder.newFolder("plugins");
File confFolder = tempFolder.newFolder("conf");
// adjust the test environment for the purposes of this test
Map<String, String> map = new HashMap<String, String>(System.getenv());
map.put(ENV_FLINK_BIN_DIR, binFolder.getAbsolutePath());
map.put(ENV_FLINK_LIB_DIR, libFolder.getAbsolutePath());
+ map.put(ENV_FLINK_PLUGINS_DIR, pluginsFolder.getAbsolutePath());
map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
CommonTestUtils.setEnv(map);
@@ -92,18 +99,24 @@ public class FlinkDistributionOverlayTest extends ContainerOverlayTestBase {
assertEquals(binFolder.getAbsolutePath(), builder.flinkBinPath.getAbsolutePath());
assertEquals(libFolder.getAbsolutePath(), builder.flinkLibPath.getAbsolutePath());
+ assertEquals(pluginsFolder.getAbsolutePath(), builder.flinkPluginsPath.getAbsolutePath());
assertEquals(confFolder.getAbsolutePath(), builder.flinkConfPath.getAbsolutePath());
}
@Test
public void testBuilderFromEnvironmentBad() throws Exception {
+ testBuilderFromEnvironmentBad(ENV_FLINK_BIN_DIR);
+ testBuilderFromEnvironmentBad(ENV_FLINK_LIB_DIR);
+ testBuilderFromEnvironmentBad(ENV_FLINK_PLUGINS_DIR);
+ testBuilderFromEnvironmentBad(ENV_FLINK_CONF_DIR);
+ }
+
+ public void testBuilderFromEnvironmentBad(String obligatoryEnvironmentVariable) throws Exception {
Configuration conf = new Configuration();
// adjust the test environment for the purposes of this test
Map<String, String> map = new HashMap<>(System.getenv());
- map.remove(ENV_FLINK_BIN_DIR);
- map.remove(ENV_FLINK_LIB_DIR);
- map.remove(ENV_FLINK_CONF_DIR);
+ map.remove(obligatoryEnvironmentVariable);
CommonTestUtils.setEnv(map);
try {
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index 3135ecf..ebf470c 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -102,6 +102,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_LIB_DIR;
+import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_PLUGINS_DIR;
import static org.apache.flink.runtime.entrypoint.component.FileJobGraphRetriever.JOB_GRAPH_FILE_PATH;
import static org.apache.flink.yarn.cli.FlinkYarnSessionCli.CONFIG_FILE_LOG4J_NAME;
import static org.apache.flink.yarn.cli.FlinkYarnSessionCli.CONFIG_FILE_LOGBACK_NAME;
@@ -700,7 +701,7 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
}
- addLibFolderToShipFiles(systemShipFiles);
+ addEnvironmentFoldersToShipFiles(systemShipFiles);
// Set-up ApplicationSubmissionContext for the application
@@ -1513,23 +1514,32 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
}
- protected void addLibFolderToShipFiles(Collection<File> effectiveShipFiles) {
+ protected void addEnvironmentFoldersToShipFiles(Collection<File> effectiveShipFiles) {
// Add lib folder to the ship files if the environment variable is set.
// This is for convenience when running from the command-line.
// (for other files users explicitly set the ship files)
String libDir = System.getenv().get(ENV_FLINK_LIB_DIR);
if (libDir != null) {
- File libDirFile = new File(libDir);
- if (libDirFile.isDirectory()) {
- effectiveShipFiles.add(libDirFile);
- } else {
- throw new YarnDeploymentException("The environment variable '" + ENV_FLINK_LIB_DIR +
- "' is set to '" + libDir + "' but the directory doesn't exist.");
- }
+ addEnvFolderToShipFiles(effectiveShipFiles, libDir, ENV_FLINK_LIB_DIR);
} else if (this.shipFiles.isEmpty()) {
LOG.warn("Environment variable '{}' not set and ship files have not been provided manually. " +
"Not shipping any library files.", ENV_FLINK_LIB_DIR);
}
+
+ String pluginsDir = System.getenv().get(ENV_FLINK_PLUGINS_DIR);
+ if (pluginsDir != null) {
+ addEnvFolderToShipFiles(effectiveShipFiles, pluginsDir, ENV_FLINK_PLUGINS_DIR);
+ }
+ }
+
+ private void addEnvFolderToShipFiles(Collection<File> effectiveShipFiles, String directory, String environmentVariableName) {
+ File directoryFile = new File(directory);
+ if (directoryFile.isDirectory()) {
+ effectiveShipFiles.add(directoryFile);
+ } else {
+ throw new YarnDeploymentException("The environment variable '" + environmentVariableName +
+ "' is set to '" + directory + "' but the directory doesn't exist.");
+ }
}
protected ContainerLaunchContext setupApplicationMasterContainer(
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 6c2e42e..1d1d44f 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -435,10 +435,10 @@ public class YarnClusterDescriptorTest extends TestLogger {
}
/**
- * Tests to ship a lib folder through the {@code YarnClusterDescriptor.addShipFiles}.
+ * Tests to ship files through the {@code YarnClusterDescriptor.addShipFiles}.
*/
@Test
- public void testExplicitLibShipping() throws Exception {
+ public void testExplicitFileShipping() throws Exception {
try (YarnClusterDescriptor descriptor = new YarnClusterDescriptor(
new Configuration(),
yarnConfiguration,
@@ -464,7 +464,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
// only execute part of the deployment to test for shipped files
Set<File> effectiveShipFiles = new HashSet<>();
- descriptor.addLibFolderToShipFiles(effectiveShipFiles);
+ descriptor.addEnvironmentFoldersToShipFiles(effectiveShipFiles);
Assert.assertEquals(0, effectiveShipFiles.size());
Assert.assertEquals(2, descriptor.shipFiles.size());
@@ -473,11 +473,17 @@ public class YarnClusterDescriptorTest extends TestLogger {
}
}
- /**
- * Tests to ship a lib folder through the {@code ConfigConstants.ENV_FLINK_LIB_DIR}.
- */
@Test
public void testEnvironmentLibShipping() throws Exception {
+ testEnvironmentDirectoryShipping(ConfigConstants.ENV_FLINK_LIB_DIR);
+ }
+
+ @Test
+ public void testEnvironmentPluginsShipping() throws Exception {
+ testEnvironmentDirectoryShipping(ConfigConstants.ENV_FLINK_PLUGINS_DIR);
+ }
+
+ public void testEnvironmentDirectoryShipping(String environmentVariable) throws Exception {
try (YarnClusterDescriptor descriptor = new YarnClusterDescriptor(
new Configuration(),
yarnConfiguration,
@@ -493,10 +499,10 @@ public class YarnClusterDescriptorTest extends TestLogger {
final Map<String, String> oldEnv = System.getenv();
try {
Map<String, String> env = new HashMap<>(1);
- env.put(ConfigConstants.ENV_FLINK_LIB_DIR, libFolder.getAbsolutePath());
+ env.put(environmentVariable, libFolder.getAbsolutePath());
CommonTestUtils.setEnv(env);
// only execute part of the deployment to test for shipped files
- descriptor.addLibFolderToShipFiles(effectiveShipFiles);
+ descriptor.addEnvironmentFoldersToShipFiles(effectiveShipFiles);
} finally {
CommonTestUtils.setEnv(oldEnv);
}
[flink] 03/10: [hotfix][runtime] Deduplicate and simplify
environment variables handling in FlinkDistributionOverlay
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6ed28c84d4b0ba95660332663a687bdccedaa699
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Apr 24 18:06:46 2019 +0200
[hotfix][runtime] Deduplicate and simplify environment variables handling in FlinkDistributionOverlay
---
.../overlays/FlinkDistributionOverlay.java | 32 +++++++---------------
1 file changed, 10 insertions(+), 22 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
index 0655554..baebd8c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/overlays/FlinkDistributionOverlay.java
@@ -33,6 +33,7 @@ import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_HOME_DIR;
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_LIB_DIR;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* Overlays Flink into a container, based on supplied bin/conf/lib directories.
@@ -93,28 +94,9 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
* @param globalConfiguration the current configuration.
*/
public Builder fromEnvironment(Configuration globalConfiguration) {
-
- Map<String,String> env = System.getenv();
- if(env.containsKey(ENV_FLINK_BIN_DIR)) {
- flinkBinPath = new File(System.getenv(ENV_FLINK_BIN_DIR));
- }
- else {
- throw new IllegalStateException(String.format("the %s environment variable must be set", ENV_FLINK_BIN_DIR));
- }
-
- if(env.containsKey(ENV_FLINK_CONF_DIR)) {
- flinkConfPath = new File(System.getenv(ENV_FLINK_CONF_DIR));
- }
- else {
- throw new IllegalStateException(String.format("the %s environment variable must be set", ENV_FLINK_CONF_DIR));
- }
-
- if(env.containsKey(ENV_FLINK_LIB_DIR)) {
- flinkLibPath = new File(System.getenv(ENV_FLINK_LIB_DIR));
- }
- else {
- throw new IllegalStateException(String.format("the %s environment variable must be set", ENV_FLINK_LIB_DIR));
- }
+ flinkBinPath = getObligatoryFileFromEnvironment(ENV_FLINK_BIN_DIR);
+ flinkConfPath = getObligatoryFileFromEnvironment(ENV_FLINK_CONF_DIR);
+ flinkLibPath = getObligatoryFileFromEnvironment(ENV_FLINK_LIB_DIR);
return this;
}
@@ -122,5 +104,11 @@ public class FlinkDistributionOverlay extends AbstractContainerOverlay {
public FlinkDistributionOverlay build() {
return new FlinkDistributionOverlay(flinkBinPath, flinkConfPath, flinkLibPath);
}
+
+ private static File getObligatoryFileFromEnvironment(String envVariableName) {
+ String directory = System.getenv(envVariableName);
+ checkState(directory != null, "the %s environment variable must be set", envVariableName);
+ return new File(directory);
+ }
}
}
[flink] 02/10: [hotfix][dist] Export variables in only one place
(config.sh)
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9871f252e683412f9b56eeba1c8013330248e75d
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Apr 24 12:11:18 2019 +0200
[hotfix][dist] Export variables in only one place (config.sh)
Previously exporting of variables like FLINK_CONF_DIR, FLINK_BIN_DIR, FLINK_LIB_DIR, ...
was scattered or even dupliacted in multiple places. Now it happens only in config.sh.
---
flink-dist/src/main/flink-bin/bin/config.sh | 14 ++++++++------
flink-dist/src/main/flink-bin/bin/flink | 3 ---
.../src/main/flink-bin/mesos-bin/mesos-appmaster-job.sh | 4 ----
flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh | 4 ----
.../src/main/flink-bin/mesos-bin/mesos-taskmanager.sh | 4 ----
flink-dist/src/main/flink-bin/yarn-bin/yarn-session.sh | 2 --
flink-table/flink-sql-client/bin/sql-client.sh | 3 ---
7 files changed, 8 insertions(+), 26 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 3af82b2..e430e91 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -300,12 +300,6 @@ FLINK_HOME=`dirname "$SYMLINK_RESOLVED_BIN"`
FLINK_LIB_DIR=$FLINK_HOME/lib
FLINK_OPT_DIR=$FLINK_HOME/opt
-### Exported environment variables ###
-export FLINK_CONF_DIR
-# export /lib dir to access it during deployment of the Yarn staging files
-export FLINK_LIB_DIR
-# export /opt dir to access it for the SQL client
-export FLINK_OPT_DIR
# These need to be mangled because they are directly passed to java.
# The above lib path is used by the shell script to retrieve jars in a
@@ -317,6 +311,14 @@ DEFAULT_FLINK_LOG_DIR=$FLINK_HOME_DIR_MANGLED/log
FLINK_CONF_FILE="flink-conf.yaml"
YAML_CONF=${FLINK_CONF_DIR}/${FLINK_CONF_FILE}
+### Exported environment variables ###
+export FLINK_CONF_DIR
+export FLINK_BIN_DIR
+# export /lib dir to access it during deployment of the Yarn staging files
+export FLINK_LIB_DIR
+# export /opt dir to access it for the SQL client
+export FLINK_OPT_DIR
+
########################################################################################################################
# ENVIRONMENT VARIABLES
########################################################################################################################
diff --git a/flink-dist/src/main/flink-bin/bin/flink b/flink-dist/src/main/flink-bin/bin/flink
index 113d4ec..9e01d4b 100644
--- a/flink-dist/src/main/flink-bin/bin/flink
+++ b/flink-dist/src/main/flink-bin/bin/flink
@@ -48,8 +48,5 @@ CC_CLASSPATH=`constructFlinkClassPath`
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-client-$HOSTNAME.log
log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
-export FLINK_HOME
-export FLINK_CONF_DIR
-
# Add HADOOP_CLASSPATH to allow the usage of Hadoop file systems
exec $JAVA_RUN $JVM_ARGS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"
diff --git a/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster-job.sh b/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster-job.sh
index b21670a..3d67d38 100755
--- a/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster-job.sh
+++ b/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster-job.sh
@@ -32,10 +32,6 @@ CC_CLASSPATH=`manglePathList $(constructFlinkClassPath):$INTERNAL_HADOOP_CLASSPA
log="${FLINK_LOG_DIR}/flink-${FLINK_IDENT_STRING}-mesos-appmaster-${HOSTNAME}.log"
log_setting="-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml"
-export FLINK_CONF_DIR
-export FLINK_BIN_DIR
-export FLINK_LIB_DIR
-
exec $JAVA_RUN $JVM_ARGS -classpath "$CC_CLASSPATH" $log_setting org.apache.flink.mesos.entrypoint.MesosJobClusterEntrypoint "$@"
rc=$?
diff --git a/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh b/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh
index f8a94b4..b49fba8 100755
--- a/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh
+++ b/flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh
@@ -32,10 +32,6 @@ CC_CLASSPATH=`manglePathList $(constructFlinkClassPath):$INTERNAL_HADOOP_CLASSPA
log="${FLINK_LOG_DIR}/flink-${FLINK_IDENT_STRING}-mesos-appmaster-${HOSTNAME}.log"
log_setting="-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml"
-export FLINK_CONF_DIR
-export FLINK_BIN_DIR
-export FLINK_LIB_DIR
-
ENTRY_POINT=org.apache.flink.mesos.entrypoint.MesosSessionClusterEntrypoint
exec $JAVA_RUN $JVM_ARGS -classpath "$CC_CLASSPATH" $log_setting ${ENTRY_POINT} "$@"
diff --git a/flink-dist/src/main/flink-bin/mesos-bin/mesos-taskmanager.sh b/flink-dist/src/main/flink-bin/mesos-bin/mesos-taskmanager.sh
index a484d09..52232ab 100755
--- a/flink-dist/src/main/flink-bin/mesos-bin/mesos-taskmanager.sh
+++ b/flink-dist/src/main/flink-bin/mesos-bin/mesos-taskmanager.sh
@@ -37,10 +37,6 @@ export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_MEM}"
# Add TaskManager-specific JVM options
export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_TM}"
-export FLINK_CONF_DIR
-export FLINK_BIN_DIR
-export FLINK_LIB_DIR
-
ENTRY_POINT=org.apache.flink.mesos.entrypoint.MesosTaskExecutorRunner
exec $JAVA_RUN $JVM_ARGS ${FLINK_ENV_JAVA_OPTS} -classpath "$CC_CLASSPATH" $log_setting ${ENTRY_POINT} "$@"
diff --git a/flink-dist/src/main/flink-bin/yarn-bin/yarn-session.sh b/flink-dist/src/main/flink-bin/yarn-bin/yarn-session.sh
index 03b1e3a..a41a1de 100755
--- a/flink-dist/src/main/flink-bin/yarn-bin/yarn-session.sh
+++ b/flink-dist/src/main/flink-bin/yarn-bin/yarn-session.sh
@@ -34,7 +34,5 @@ CC_CLASSPATH=`manglePathList $(constructFlinkClassPath):$INTERNAL_HADOOP_CLASSPA
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-yarn-session-$HOSTNAME.log
log_setting="-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-yarn-session.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback-yarn.xml"
-export FLINK_CONF_DIR
-
$JAVA_RUN $JVM_ARGS -classpath "$CC_CLASSPATH" $log_setting org.apache.flink.yarn.cli.FlinkYarnSessionCli -j "$FLINK_LIB_DIR"/flink-dist*.jar "$@"
diff --git a/flink-table/flink-sql-client/bin/sql-client.sh b/flink-table/flink-sql-client/bin/sql-client.sh
index 817d9e6..9ba20bd 100644
--- a/flink-table/flink-sql-client/bin/sql-client.sh
+++ b/flink-table/flink-sql-client/bin/sql-client.sh
@@ -56,9 +56,6 @@ CC_CLASSPATH=`constructFlinkClassPath`
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-sql-client-$HOSTNAME.log
log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
-export FLINK_HOME
-export FLINK_CONF_DIR
-
# get path of jar in /opt if it exist
FLINK_SQL_CLIENT_JAR=$(find "$FLINK_OPT_DIR" -regex ".*flink-sql-client.*.jar")
[flink] 10/10: [FLINK-12143][e2e] Change end-to-end tests with file
system to use plugins mechanism
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9e6ff81e22d6f5f04abb50ca1aea84fd2542bf9d
Author: Aleksey Pak <al...@ververica.com>
AuthorDate: Thu Jun 6 12:42:21 2019 +0200
[FLINK-12143][e2e] Change end-to-end tests with file system to use plugins mechanism
---
.../test-scripts/common_dummy_fs.sh | 23 ++++++++++++++++++++++
flink-end-to-end-tests/test-scripts/common_s3.sh | 2 +-
.../test-scripts/test_azure_fs.sh | 2 +-
.../test-scripts/test_batch_wordcount.sh | 3 ++-
.../test-scripts/test_docker_embedded_job.sh | 3 ++-
.../test-scripts/test_yarn_kerberos_docker.sh | 3 ++-
6 files changed, 31 insertions(+), 5 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh b/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh
new file mode 100644
index 0000000..bd6e756
--- /dev/null
+++ b/flink-end-to-end-tests/test-scripts/common_dummy_fs.sh
@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+################################################################################
+# 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.
+################################################################################
+
+function dummy_fs_setup() {
+ mkdir -p "$FLINK_DIR/plugins/dummy-fs"
+ cp "${END_TO_END_DIR}/flink-plugins-test/target/flink-dummy-fs.jar" "${FLINK_DIR}/plugins/dummy-fs/"
+}
diff --git a/flink-end-to-end-tests/test-scripts/common_s3.sh b/flink-end-to-end-tests/test-scripts/common_s3.sh
index 6536a88..b423977 100644
--- a/flink-end-to-end-tests/test-scripts/common_s3.sh
+++ b/flink-end-to-end-tests/test-scripts/common_s3.sh
@@ -63,7 +63,7 @@ s3util="java -jar ${END_TO_END_DIR}/flink-e2e-test-utils/target/S3UtilProgram.ja
# None
###################################
function s3_setup {
- add_optional_lib "s3-fs-$1"
+ add_optional_plugin "s3-fs-$1"
set_config_key "s3.access-key" "$IT_CASE_S3_ACCESS_KEY"
set_config_key "s3.secret-key" "$IT_CASE_S3_SECRET_KEY"
}
diff --git a/flink-end-to-end-tests/test-scripts/test_azure_fs.sh b/flink-end-to-end-tests/test-scripts/test_azure_fs.sh
index f68c6d8..b200e49 100755
--- a/flink-end-to-end-tests/test-scripts/test_azure_fs.sh
+++ b/flink-end-to-end-tests/test-scripts/test_azure_fs.sh
@@ -61,7 +61,7 @@ AZURE_TEST_DATA_WORDS_URI="wasbs://$IT_CASE_AZURE_CONTAINER@$IT_CASE_AZURE_ACCOU
function azure_setup {
echo "Copying flink azure jars and writing out configs"
- add_optional_lib "azure-fs-hadoop"
+ add_optional_plugin "azure-fs-hadoop"
set_config_key "fs.azure.account.key.$IT_CASE_AZURE_ACCOUNT.blob.core.windows.net" "$IT_CASE_AZURE_ACCESS_KEY"
}
diff --git a/flink-end-to-end-tests/test-scripts/test_batch_wordcount.sh b/flink-end-to-end-tests/test-scripts/test_batch_wordcount.sh
index 3b8400b..53bd585 100755
--- a/flink-end-to-end-tests/test-scripts/test_batch_wordcount.sh
+++ b/flink-end-to-end-tests/test-scripts/test_batch_wordcount.sh
@@ -35,7 +35,8 @@ case $INPUT_TYPE in
INPUT_LOCATION="${S3_TEST_DATA_WORDS_URI}"
;;
(dummy-fs)
- cp "${END_TO_END_DIR}/flink-plugins-test/target/flink-dummy-fs.jar" "${FLINK_DIR}/lib/"
+ source "$(dirname "$0")"/common_dummy_fs.sh
+ dummy_fs_setup
INPUT_LOCATION="dummy://localhost/words"
;;
(*)
diff --git a/flink-end-to-end-tests/test-scripts/test_docker_embedded_job.sh b/flink-end-to-end-tests/test-scripts/test_docker_embedded_job.sh
index dabb3ce..4b87d57 100755
--- a/flink-end-to-end-tests/test-scripts/test_docker_embedded_job.sh
+++ b/flink-end-to-end-tests/test-scripts/test_docker_embedded_job.sh
@@ -37,7 +37,8 @@ case $INPUT_TYPE in
INPUT_LOCATION=${INPUT_PATH}/words
;;
(dummy-fs)
- cp "${END_TO_END_DIR}/flink-plugins-test/target/flink-dummy-fs.jar" "${FLINK_DIR}/lib/"
+ source "$(dirname "$0")"/common_dummy_fs.sh
+ dummy_fs_setup
INPUT_LOCATION="dummy://localhost/words"
;;
(*)
diff --git a/flink-end-to-end-tests/test-scripts/test_yarn_kerberos_docker.sh b/flink-end-to-end-tests/test-scripts/test_yarn_kerberos_docker.sh
index 5199b2a..6f66236 100755
--- a/flink-end-to-end-tests/test-scripts/test_yarn_kerberos_docker.sh
+++ b/flink-end-to-end-tests/test-scripts/test_yarn_kerberos_docker.sh
@@ -43,7 +43,8 @@ case $INPUT_TYPE in
EXPECTED_RESULT_LOG_CONTAINS=("consummation,1" "of,14" "calamity,1")
;;
(dummy-fs)
- cp "${END_TO_END_DIR}/flink-plugins-test/target/flink-dummy-fs.jar" "${FLINK_DIR}/lib/"
+ source "$(dirname "$0")"/common_dummy_fs.sh
+ dummy_fs_setup
INPUT_ARGS="--input dummy://localhost/words"
EXPECTED_RESULT_LOG_CONTAINS=("my,1" "dear,2" "world,2")
;;
[flink] 01/10: [hotfix][dist] Rename FLINK_ROOT*_DIR to FLINK_HOME
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d14117f651b5753e36d777f3ab50dc8559f943bf
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Apr 24 11:43:08 2019 +0200
[hotfix][dist] Rename FLINK_ROOT*_DIR to FLINK_HOME
The same concept of Flink's root/home directory was named in different places differently.
This commit unifies the naming convention to FLINK_HOME by renaming FLINK_ROOT_DIR and FLINK_HOME_DIR to FLINK_HOME.
---
flink-dist/src/main/flink-bin/bin/config.sh | 14 +++++++-------
flink-dist/src/main/flink-bin/bin/flink | 2 +-
flink-dist/src/main/flink-bin/bin/flink.bat | 4 ++--
.../src/main/flink-bin/bin/pyflink-gateway-server.sh | 4 ++--
flink-dist/src/main/flink-bin/bin/pyflink-stream.sh | 2 +-
flink-dist/src/main/flink-bin/bin/pyflink.bat | 4 ++--
flink-dist/src/main/flink-bin/bin/pyflink.sh | 2 +-
flink-dist/src/main/flink-bin/bin/start-cluster.bat | 8 ++++----
flink-table/flink-sql-client/bin/sql-client.sh | 2 +-
9 files changed, 21 insertions(+), 21 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 1d38e2a..3af82b2 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -296,9 +296,9 @@ bin=`dirname "$target"`
SYMLINK_RESOLVED_BIN=`cd "$bin"; pwd -P`
# Define the main directory of the flink installation
-FLINK_ROOT_DIR=`dirname "$SYMLINK_RESOLVED_BIN"`
-FLINK_LIB_DIR=$FLINK_ROOT_DIR/lib
-FLINK_OPT_DIR=$FLINK_ROOT_DIR/opt
+FLINK_HOME=`dirname "$SYMLINK_RESOLVED_BIN"`
+FLINK_LIB_DIR=$FLINK_HOME/lib
+FLINK_OPT_DIR=$FLINK_HOME/opt
### Exported environment variables ###
export FLINK_CONF_DIR
@@ -310,10 +310,10 @@ export FLINK_OPT_DIR
# These need to be mangled because they are directly passed to java.
# The above lib path is used by the shell script to retrieve jars in a
# directory, so it needs to be unmangled.
-FLINK_ROOT_DIR_MANGLED=`manglePath "$FLINK_ROOT_DIR"`
-if [ -z "$FLINK_CONF_DIR" ]; then FLINK_CONF_DIR=$FLINK_ROOT_DIR_MANGLED/conf; fi
-FLINK_BIN_DIR=$FLINK_ROOT_DIR_MANGLED/bin
-DEFAULT_FLINK_LOG_DIR=$FLINK_ROOT_DIR_MANGLED/log
+FLINK_HOME_DIR_MANGLED=`manglePath "$FLINK_HOME"`
+if [ -z "$FLINK_CONF_DIR" ]; then FLINK_CONF_DIR=$FLINK_HOME_DIR_MANGLED/conf; fi
+FLINK_BIN_DIR=$FLINK_HOME_DIR_MANGLED/bin
+DEFAULT_FLINK_LOG_DIR=$FLINK_HOME_DIR_MANGLED/log
FLINK_CONF_FILE="flink-conf.yaml"
YAML_CONF=${FLINK_CONF_DIR}/${FLINK_CONF_FILE}
diff --git a/flink-dist/src/main/flink-bin/bin/flink b/flink-dist/src/main/flink-bin/bin/flink
index d38217d..113d4ec 100644
--- a/flink-dist/src/main/flink-bin/bin/flink
+++ b/flink-dist/src/main/flink-bin/bin/flink
@@ -48,7 +48,7 @@ CC_CLASSPATH=`constructFlinkClassPath`
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-client-$HOSTNAME.log
log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
-export FLINK_ROOT_DIR
+export FLINK_HOME
export FLINK_CONF_DIR
# Add HADOOP_CLASSPATH to allow the usage of Hadoop file systems
diff --git a/flink-dist/src/main/flink-bin/bin/flink.bat b/flink-dist/src/main/flink-bin/bin/flink.bat
index 845d791..8ed1fcf 100644
--- a/flink-dist/src/main/flink-bin/bin/flink.bat
+++ b/flink-dist/src/main/flink-bin/bin/flink.bat
@@ -20,8 +20,8 @@
setlocal
SET bin=%~dp0
-SET FLINK_ROOT_DIR=%bin%..
-SET FLINK_LIB_DIR=%FLINK_ROOT_DIR%\lib
+SET FLINK_HOME=%bin%..
+SET FLINK_LIB_DIR=%FLINK_HOME%\lib
SET JVM_ARGS=-Xmx512m
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink-gateway-server.sh b/flink-dist/src/main/flink-bin/bin/pyflink-gateway-server.sh
index 5fff644..433c626 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink-gateway-server.sh
+++ b/flink-dist/src/main/flink-bin/bin/pyflink-gateway-server.sh
@@ -50,8 +50,8 @@ done
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-python-$HOSTNAME.log
log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
-TABLE_JAR_PATH=`echo "$FLINK_ROOT_DIR"/opt/flink-table*.jar`
-PYTHON_JAR_PATH=`echo "$FLINK_ROOT_DIR"/opt/flink-python*java-binding.jar`
+TABLE_JAR_PATH=`echo "$FLINK_HOME"/opt/flink-table*.jar`
+PYTHON_JAR_PATH=`echo "$FLINK_HOME"/opt/flink-python*java-binding.jar`
FLINK_TEST_CLASSPATH=""
if [[ -n "$FLINK_TESTING" ]]; then
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh b/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
index 3e44a10..5e1b1ae 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
+++ b/flink-dist/src/main/flink-bin/bin/pyflink-stream.sh
@@ -22,4 +22,4 @@ bin=`cd "$bin"; pwd`
. "$bin"/config.sh
-"$FLINK_BIN_DIR"/flink run --class org.apache.flink.streaming.python.api.PythonStreamBinder -v "$FLINK_ROOT_DIR"/opt/flink-streaming-python*.jar "$@"
+"$FLINK_BIN_DIR"/flink run --class org.apache.flink.streaming.python.api.PythonStreamBinder -v "$FLINK_HOME"/opt/flink-streaming-python*.jar "$@"
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink.bat b/flink-dist/src/main/flink-bin/bin/pyflink.bat
index 4d7fa0e..65c4f21 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink.bat
+++ b/flink-dist/src/main/flink-bin/bin/pyflink.bat
@@ -20,6 +20,6 @@
setlocal EnableDelayedExpansion
SET bin=%~dp0
-SET FLINK_ROOT_DIR=%bin%..
+SET FLINK_HOME=%bin%..
-"%FLINK_ROOT_DIR%\bin\flink" run -v "%FLINK_ROOT_DIR%"\opt\flink-python*.jar %*
+"%FLINK_HOME%\bin\flink" run -v "%FLINK_HOME%"\opt\flink-python*.jar %*
diff --git a/flink-dist/src/main/flink-bin/bin/pyflink.sh b/flink-dist/src/main/flink-bin/bin/pyflink.sh
index 46470e8..8f7a55c 100644
--- a/flink-dist/src/main/flink-bin/bin/pyflink.sh
+++ b/flink-dist/src/main/flink-bin/bin/pyflink.sh
@@ -22,4 +22,4 @@ bin=`cd "$bin"; pwd`
. "$bin"/config.sh
-"$FLINK_BIN_DIR"/flink run -v "$FLINK_ROOT_DIR"/opt/flink-python_*.jar "$@"
+"$FLINK_BIN_DIR"/flink run -v "$FLINK_HOME"/opt/flink-python_*.jar "$@"
diff --git a/flink-dist/src/main/flink-bin/bin/start-cluster.bat b/flink-dist/src/main/flink-bin/bin/start-cluster.bat
index eb94d31..a852a55 100644
--- a/flink-dist/src/main/flink-bin/bin/start-cluster.bat
+++ b/flink-dist/src/main/flink-bin/bin/start-cluster.bat
@@ -20,10 +20,10 @@
setlocal EnableDelayedExpansion
SET bin=%~dp0
-SET FLINK_ROOT_DIR=%bin%..
-SET FLINK_LIB_DIR=%FLINK_ROOT_DIR%\lib
-SET FLINK_CONF_DIR=%FLINK_ROOT_DIR%\conf
-SET FLINK_LOG_DIR=%FLINK_ROOT_DIR%\log
+SET FLINK_HOME=%bin%..
+SET FLINK_LIB_DIR=%FLINK_HOME%\lib
+SET FLINK_CONF_DIR=%FLINK_HOME%\conf
+SET FLINK_LOG_DIR=%FLINK_HOME%\log
SET JVM_ARGS=-Xms1024m -Xmx1024m
diff --git a/flink-table/flink-sql-client/bin/sql-client.sh b/flink-table/flink-sql-client/bin/sql-client.sh
index e4e9315..817d9e6 100644
--- a/flink-table/flink-sql-client/bin/sql-client.sh
+++ b/flink-table/flink-sql-client/bin/sql-client.sh
@@ -56,7 +56,7 @@ CC_CLASSPATH=`constructFlinkClassPath`
log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-sql-client-$HOSTNAME.log
log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
-export FLINK_ROOT_DIR
+export FLINK_HOME
export FLINK_CONF_DIR
# get path of jar in /opt if it exist
[flink] 08/10: [FLINK-12143] PluginLoader: use system level always
parents first patterns in class loading
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 33b182cfa946cfe98459207d0c428afbfd041cca
Author: Aleksey Pak <al...@ververica.com>
AuthorDate: Fri May 10 14:30:45 2019 +0200
[FLINK-12143] PluginLoader: use system level always parents first patterns in class loading
---
.../apache/flink/configuration/CoreOptions.java | 8 +---
.../org/apache/flink/core/plugin/PluginConfig.java | 28 ++++++++----
.../org/apache/flink/core/plugin/PluginLoader.java | 5 +-
.../apache/flink/core/plugin/PluginManager.java | 14 ++++--
.../org/apache/flink/core/plugin/PluginUtils.java | 4 +-
.../java/org/apache/flink/util/ArrayUtils.java | 41 +++++++++++++++++
.../java/org/apache/flink/util/ArrayUtilsTest.java | 53 ++++++++++++++++++++++
.../apache/flink/test/plugin/PluginLoaderTest.java | 6 +--
.../flink/test/plugin/PluginManagerTest.java | 2 +-
9 files changed, 134 insertions(+), 27 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
index a70c020..6132284 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.docs.ConfigGroup;
import org.apache.flink.annotation.docs.ConfigGroups;
import org.apache.flink.annotation.docs.Documentation;
import org.apache.flink.configuration.description.Description;
+import org.apache.flink.util.ArrayUtils;
import static org.apache.flink.configuration.ConfigOptions.key;
@@ -116,12 +117,7 @@ public class CoreOptions {
if (append.isEmpty()) {
return basePatterns;
} else {
- String[] appendPatterns = append.split(";");
-
- String[] joinedPatterns = new String[basePatterns.length + appendPatterns.length];
- System.arraycopy(basePatterns, 0, joinedPatterns, 0, basePatterns.length);
- System.arraycopy(appendPatterns, 0, joinedPatterns, basePatterns.length, appendPatterns.length);
- return joinedPatterns;
+ return ArrayUtils.concat(basePatterns, append.split(";"));
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
index c4a3b2b..b90c87c 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginConfig.java
@@ -20,6 +20,7 @@ package org.apache.flink.core.plugin;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
import java.io.File;
import java.nio.file.Path;
@@ -28,31 +29,40 @@ import java.util.Optional;
/**
* Stores the configuration for plugins mechanism.
*/
+@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
public class PluginConfig {
private final Optional<Path> pluginsPath;
- private PluginConfig() {
- this.pluginsPath = Optional.empty();
- }
+ private final String[] alwaysParentFirstPatterns;
- private PluginConfig(Path pluginsPath) {
- this.pluginsPath = Optional.of(pluginsPath);
+ private PluginConfig(Optional<Path> pluginsPath, String[] alwaysParentFirstPatterns) {
+ this.pluginsPath = pluginsPath;
+ this.alwaysParentFirstPatterns = alwaysParentFirstPatterns;
}
public Optional<Path> getPluginsPath() {
return pluginsPath;
}
+ public String[] getAlwaysParentFirstPatterns() {
+ return alwaysParentFirstPatterns;
+ }
+
public static PluginConfig fromConfiguration(Configuration configuration) {
+ return new PluginConfig(
+ getPluginsDirPath(configuration),
+ CoreOptions.getParentFirstLoaderPatterns(configuration));
+ }
+
+ private static Optional<Path> getPluginsDirPath(Configuration configuration) {
String pluginsDir = configuration.getString(ConfigConstants.ENV_FLINK_PLUGINS_DIR, null);
if (pluginsDir == null) {
- return new PluginConfig();
+ return Optional.empty();
}
-
File pluginsDirFile = new File(pluginsDir);
if (!pluginsDirFile.isDirectory()) {
- return new PluginConfig();
+ return Optional.empty();
}
- return new PluginConfig(pluginsDirFile.toPath());
+ return Optional.of(pluginsDirFile.toPath());
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
index 70468bf..f90bda8 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginLoader.java
@@ -19,6 +19,7 @@
package org.apache.flink.core.plugin;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.ArrayUtils;
import org.apache.flink.util.ChildFirstClassLoader;
import javax.annotation.concurrent.ThreadSafe;
@@ -39,12 +40,12 @@ public class PluginLoader {
private final ClassLoader pluginClassLoader;
@VisibleForTesting
- public PluginLoader(PluginDescriptor pluginDescriptor, ClassLoader parentClassLoader) {
+ public PluginLoader(PluginDescriptor pluginDescriptor, ClassLoader parentClassLoader, String[] alwaysParentFirstPatterns) {
this.pluginClassLoader =
new ChildFirstClassLoader(
pluginDescriptor.getPluginResourceURLs(),
parentClassLoader,
- pluginDescriptor.getLoaderExcludePatterns());
+ ArrayUtils.concat(alwaysParentFirstPatterns, pluginDescriptor.getLoaderExcludePatterns()));
}
/**
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
index 00971fa..c6835a2 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginManager.java
@@ -25,6 +25,7 @@ import org.apache.flink.shaded.guava18.com.google.common.collect.Iterators;
import javax.annotation.concurrent.ThreadSafe;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
@@ -41,13 +42,17 @@ public class PluginManager {
/** A collection of descriptions of all plugins known to this plugin manager. */
private final Collection<PluginDescriptor> pluginDescriptors;
- public PluginManager(Collection<PluginDescriptor> pluginDescriptors) {
- this(pluginDescriptors, PluginManager.class.getClassLoader());
+ /** List of patterns for classes that should always be resolved from the parent ClassLoader. */
+ private final String[] alwaysParentFirstPatterns;
+
+ public PluginManager(Collection<PluginDescriptor> pluginDescriptors, String[] alwaysParentFirstPatterns) {
+ this(pluginDescriptors, PluginManager.class.getClassLoader(), alwaysParentFirstPatterns);
}
- public PluginManager(Collection<PluginDescriptor> pluginDescriptors, ClassLoader parentClassLoader) {
+ public PluginManager(Collection<PluginDescriptor> pluginDescriptors, ClassLoader parentClassLoader, String[] alwaysParentFirstPatterns) {
this.pluginDescriptors = pluginDescriptors;
this.parentClassLoader = parentClassLoader;
+ this.alwaysParentFirstPatterns = alwaysParentFirstPatterns;
}
/**
@@ -61,7 +66,7 @@ public class PluginManager {
public <P extends Plugin> Iterator<P> load(Class<P> service) {
ArrayList<Iterator<P>> combinedIterators = new ArrayList<>(pluginDescriptors.size());
for (PluginDescriptor pluginDescriptor : pluginDescriptors) {
- PluginLoader pluginLoader = new PluginLoader(pluginDescriptor, parentClassLoader);
+ PluginLoader pluginLoader = new PluginLoader(pluginDescriptor, parentClassLoader, alwaysParentFirstPatterns);
combinedIterators.add(pluginLoader.load(service));
}
return Iterators.concat(combinedIterators.iterator());
@@ -72,6 +77,7 @@ public class PluginManager {
return "PluginManager{" +
"parentClassLoader=" + parentClassLoader +
", pluginDescriptors=" + pluginDescriptors +
+ ", alwaysParentFirstPatterns=" + Arrays.toString(alwaysParentFirstPatterns) +
'}';
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
index 0f9f0d3..b6500bf 100644
--- a/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/PluginUtils.java
@@ -43,13 +43,13 @@ public final class PluginUtils {
try {
Collection<PluginDescriptor> pluginDescriptors =
new DirectoryBasedPluginFinder(pluginConfig.getPluginsPath().get()).findPlugins();
- return new PluginManager(pluginDescriptors);
+ return new PluginManager(pluginDescriptors, pluginConfig.getAlwaysParentFirstPatterns());
} catch (IOException e) {
throw new FlinkRuntimeException("Exception when trying to initialize plugin system.", e);
}
}
else {
- return new PluginManager(Collections.emptyList());
+ return new PluginManager(Collections.emptyList(), pluginConfig.getAlwaysParentFirstPatterns());
}
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/util/ArrayUtils.java b/flink-core/src/main/java/org/apache/flink/util/ArrayUtils.java
new file mode 100644
index 0000000..9877dba
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/ArrayUtils.java
@@ -0,0 +1,41 @@
+/*
+ * 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.util;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * Utility class for Java arrays.
+ */
+@Internal
+public final class ArrayUtils {
+
+ public static String[] concat(String[] array1, String[] array2) {
+ if (array1.length == 0) {
+ return array2;
+ }
+ if (array2.length == 0) {
+ return array1;
+ }
+ String[] resultArray = new String[array1.length + array2.length];
+ System.arraycopy(array1, 0, resultArray, 0, array1.length);
+ System.arraycopy(array2, 0, resultArray, array1.length, array2.length);
+ return resultArray;
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/util/ArrayUtilsTest.java b/flink-core/src/test/java/org/apache/flink/util/ArrayUtilsTest.java
new file mode 100644
index 0000000..775bae5
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/util/ArrayUtilsTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.util;
+
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link ArrayUtils}.
+ */
+public class ArrayUtilsTest extends TestLogger {
+
+ @Test
+ public void concatWithEmptyArray() {
+ String[] emptyArray = new String[] {};
+ String[] nonEmptyArray = new String[] {"some value"};
+
+ assertThat("Should return the non empty array",
+ ArrayUtils.concat(emptyArray, nonEmptyArray), sameInstance(nonEmptyArray));
+
+ assertThat("Should return the non empty array",
+ ArrayUtils.concat(nonEmptyArray, emptyArray), sameInstance(nonEmptyArray));
+ }
+
+ @Test
+ public void concatArrays() {
+ String[] array1 = new String[] {"A", "B", "C", "D", "E", "F", "G"};
+ String[] array2 = new String[] {"1", "2", "3"};
+
+ assertThat(ArrayUtils.concat(array1, array2), is(new String[]{"A", "B", "C", "D", "E", "F", "G", "1", "2", "3"}));
+
+ assertThat(ArrayUtils.concat(array2, array1), is(new String[]{"1", "2", "3", "A", "B", "C", "D", "E", "F", "G"}));
+ }
+}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
index 2460ad9..52bad19 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginLoaderTest.java
@@ -31,7 +31,7 @@ import java.util.Iterator;
/**
* Test for {@link PluginLoader}.
*/
-public class PluginLoaderTest extends PluginTestBase{
+public class PluginLoaderTest extends PluginTestBase {
@Test
public void testPluginLoading() throws Exception {
@@ -39,7 +39,7 @@ public class PluginLoaderTest extends PluginTestBase{
final URL classpathA = createPluginJarURLFromString(PLUGIN_A);
PluginDescriptor pluginDescriptorA = new PluginDescriptor("A", new URL[]{classpathA}, new String[0]);
- final PluginLoader pluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER);
+ final PluginLoader pluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
Iterator<TestSpi> testSpiIteratorA = pluginLoaderA.load(TestSpi.class);
@@ -59,7 +59,7 @@ public class PluginLoaderTest extends PluginTestBase{
Assert.assertFalse(testSpiA instanceof TestServiceA);
// In the following we check for isolation of classes between different plugin loaders.
- final PluginLoader secondPluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER);
+ final PluginLoader secondPluginLoaderA = new PluginLoader(pluginDescriptorA, PARENT_CLASS_LOADER, new String[0]);
TestSpi secondTestSpiA = secondPluginLoaderA.load(TestSpi.class).next();
Assert.assertNotNull(secondTestSpiA.testMethod());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginManagerTest.java b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginManagerTest.java
index af5667c..7439da1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginManagerTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/plugin/PluginManagerTest.java
@@ -83,7 +83,7 @@ public class PluginManagerTest extends PluginTestBase {
@Test
public void testPluginLoading() {
- final PluginManager pluginManager = new PluginManager(descriptors, PARENT_CLASS_LOADER);
+ final PluginManager pluginManager = new PluginManager(descriptors, PARENT_CLASS_LOADER, new String[0]);
final List<TestSpi> serviceImplList = Lists.newArrayList(pluginManager.load(TestSpi.class));
Assert.assertEquals(2, serviceImplList.size());