You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by pr...@apache.org on 2017/07/28 18:31:12 UTC
[apex-core] branch master updated: APEXCORE-764 Refactor Plugin
locator service
This is an automated email from the ASF dual-hosted git repository.
pramod pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apex-core.git
The following commit(s) were added to refs/heads/master by this push:
new fc0cf71 APEXCORE-764 Refactor Plugin locator service
fc0cf71 is described below
commit fc0cf71f90efc796229cc275033efa4ba85758b2
Author: Vlad Rozov <vr...@apache.org>
AuthorDate: Tue Jul 18 08:32:08 2017 -0700
APEXCORE-764 Refactor Plugin locator service
---
.../apex/engine/api/plugin/PluginLocator.java | 16 +++++---
.../plugin/AbstractApexPluginDispatcher.java | 14 ++++---
.../engine/plugin/DefaultApexPluginDispatcher.java | 29 ++++++++++-----
.../plugin/loaders/ChainedPluginLocator.java | 43 +++++++++++++++-------
.../plugin/loaders/PropertyBasedPluginLocator.java | 29 +++++++++------
.../loaders/ServiceLoaderBasedPluginLocator.java | 12 +++---
.../engine/plugin/loaders/StaticPluginLocator.java | 17 +++++----
7 files changed, 102 insertions(+), 58 deletions(-)
diff --git a/engine/src/main/java/org/apache/apex/engine/api/plugin/PluginLocator.java b/engine/src/main/java/org/apache/apex/engine/api/plugin/PluginLocator.java
index ec1c92d..9314870 100644
--- a/engine/src/main/java/org/apache/apex/engine/api/plugin/PluginLocator.java
+++ b/engine/src/main/java/org/apache/apex/engine/api/plugin/PluginLocator.java
@@ -18,22 +18,26 @@
*/
package org.apache.apex.engine.api.plugin;
-import java.util.Collection;
+import java.util.Set;
+import javax.annotation.Nonnull;
+
+import org.apache.apex.api.plugin.Plugin;
import org.apache.hadoop.conf.Configuration;
/**
- * Interface to discover plugins during Apex Master initialization. This should return collection of
+ * Interface to discover plugins during Apex Master initialization. This should return set of
* objects implementing an interface of type T.
*
* @since 3.6.0
*/
-public interface PluginLocator<T>
+public interface PluginLocator<T extends Plugin>
{
/**
- * Discover list of apex plugins.
+ * Discover apex plugins.
*
- * @return list of apex plugins.
+ * @return set of apex plugins.
*/
- Collection<T> discoverPlugins(Configuration conf);
+ @Nonnull
+ Set<T> discoverPlugins(Configuration conf);
}
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/AbstractApexPluginDispatcher.java b/engine/src/main/java/org/apache/apex/engine/plugin/AbstractApexPluginDispatcher.java
index a4aca46..cd18a5f 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/AbstractApexPluginDispatcher.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/AbstractApexPluginDispatcher.java
@@ -62,21 +62,21 @@ public abstract class AbstractApexPluginDispatcher extends AbstractService imple
protected final Collection<DAGExecutionPlugin> plugins = Lists.newArrayList();
protected final StramAppContext appContext;
protected final StreamingContainerManager dmgr;
- private final PluginLocator locator;
+ private final PluginLocator<DAGExecutionPlugin> locator;
private final AppInfo.AppStats stats;
protected Configuration launchConfig;
protected FileContext fileContext;
protected final Table<DAGExecutionEvent.Type, DAGExecutionPlugin, EventHandler<DAGExecutionEvent>> table = HashBasedTable.create();
private volatile DAG clonedDAG = null;
- public AbstractApexPluginDispatcher(PluginLocator locator, StramAppContext context, StreamingContainerManager dmgr, AppInfo.AppStats stats)
+ protected AbstractApexPluginDispatcher(String name, PluginLocator<DAGExecutionPlugin> locator, StramAppContext context, StreamingContainerManager dmgr, AppInfo.AppStats stats)
{
- super(AbstractApexPluginDispatcher.class.getName());
+ super(name);
this.locator = locator;
this.appContext = context;
this.dmgr = dmgr;
this.stats = stats;
- LOG.debug("Creating apex service ");
+ LOG.debug("Creating Plugin Dispatcher service {}", name);
}
private Configuration readLaunchConfiguration() throws IOException
@@ -122,7 +122,11 @@ public abstract class AbstractApexPluginDispatcher extends AbstractService imple
protected void serviceStop() throws Exception
{
for (DAGExecutionPlugin plugin : plugins) {
- plugin.teardown();
+ try {
+ plugin.teardown();
+ } catch (Exception e) {
+ LOG.warn("Exception during {} teardown", plugin, e);
+ }
}
super.serviceStop();
}
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/DefaultApexPluginDispatcher.java b/engine/src/main/java/org/apache/apex/engine/plugin/DefaultApexPluginDispatcher.java
index b7bbc48..b3a3855 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/DefaultApexPluginDispatcher.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/DefaultApexPluginDispatcher.java
@@ -49,13 +49,15 @@ import com.datatorrent.stram.webapp.AppInfo;
public class DefaultApexPluginDispatcher extends AbstractApexPluginDispatcher
{
private static final Logger LOG = LoggerFactory.getLogger(DefaultApexPluginDispatcher.class);
- private int qsize = 4098;
+ private static final int TIMEOUT = 10;
+ private static final int QUEUE_SIZE = 4098;
+
private ArrayBlockingQueue<Runnable> blockingQueue;
private ExecutorService executorService;
public DefaultApexPluginDispatcher(PluginLocator locator, StramAppContext context, StreamingContainerManager dmgr, AppInfo.AppStats stats)
{
- super(locator, context, dmgr, stats);
+ super(DefaultApexPluginDispatcher.class.getName(), locator, context, dmgr, stats);
}
@Override
@@ -70,8 +72,8 @@ public class DefaultApexPluginDispatcher extends AbstractApexPluginDispatcher
protected void serviceInit(Configuration conf) throws Exception
{
super.serviceInit(conf);
- LOG.debug("Creating plugin dispatch queue with size {}", qsize);
- blockingQueue = new ArrayBlockingQueue<>(qsize);
+ LOG.debug("Creating plugin dispatch queue with size {}", QUEUE_SIZE);
+ blockingQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
RejectedExecutionHandler rejectionHandler = new RejectedExecutionHandler()
{
@Override
@@ -87,18 +89,25 @@ public class DefaultApexPluginDispatcher extends AbstractApexPluginDispatcher
};
executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
- blockingQueue, new NameableThreadFactory("PluginExecutorThread"), rejectionHandler);
+ blockingQueue, new NameableThreadFactory("PluginExecutorThread", true), rejectionHandler);
}
@Override
protected void serviceStop() throws Exception
{
- executorService.shutdownNow();
- executorService.awaitTermination(10, TimeUnit.SECONDS);
- if (!executorService.isTerminated()) {
- LOG.warn("Executor service still active for plugins");
+ if (executorService != null) {
+ executorService.shutdown();
+ boolean terminated = false;
+ try {
+ terminated = executorService.awaitTermination(TIMEOUT, TimeUnit.SECONDS);
+ } finally {
+ if (!terminated) {
+ LOG.warn("{} executor service {} failed to terminate withing {} seconds", getClass().getSimpleName(), executorService, TIMEOUT);
+ }
+ executorService = null;
+ super.serviceStop();
+ }
}
- executorService = null;
}
private class ProcessEventTask<T extends DAGExecutionEvent.Type> implements Runnable
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ChainedPluginLocator.java b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ChainedPluginLocator.java
index 362928d..29c6b28 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ChainedPluginLocator.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ChainedPluginLocator.java
@@ -18,41 +18,58 @@
*/
package org.apache.apex.engine.plugin.loaders;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.apex.api.plugin.Plugin;
import org.apache.apex.engine.api.plugin.PluginLocator;
import org.apache.hadoop.conf.Configuration;
/**
* @since 3.6.0
*/
-public class ChainedPluginLocator<T> implements PluginLocator<T>
+public class ChainedPluginLocator<T extends Plugin> implements PluginLocator<T>
{
private static final Logger LOG = LoggerFactory.getLogger(ChainedPluginLocator.class);
- List<PluginLocator> locators = new ArrayList<>();
+
+ private final Set<PluginLocator<T>> locators;
public ChainedPluginLocator(PluginLocator<T>... locators)
{
- for (PluginLocator locator : locators) {
- this.locators.add(locator);
+ Collections.addAll(this.locators = new LinkedHashSet(locators.length), locators);
+ }
+
+ private static <T extends Plugin> Set<T> merge(Set<T> to, Set<T> from)
+ {
+ if (from == null || from.isEmpty()) {
+ return to;
+ }
+
+ if (to == null || to.isEmpty()) {
+ if (from instanceof LinkedHashSet) {
+ return from;
+ } else {
+ to = new LinkedHashSet<>(from.size());
+ }
}
+ to.addAll(from);
+ return to;
}
@Override
- public Collection<T> discoverPlugins(Configuration conf)
+ public Set<T> discoverPlugins(Configuration conf)
{
- List<T> plugins = new ArrayList<>();
+ Set<T> plugins = Collections.emptySet();
for (PluginLocator<T> locator : locators) {
- Collection<T> currentPlugins = locator.discoverPlugins(conf);
- if (currentPlugins != null) {
- LOG.info("Loader {} detected {} plugins ", locator.getClass().getName(), currentPlugins.size());
- plugins.addAll(currentPlugins);
+ Set<T> currentPlugins = locator.discoverPlugins(conf);
+ if (!currentPlugins.isEmpty()) {
+ LOG.info("Plugin locator {} detected {} plugins", locator.getClass().getSimpleName(), currentPlugins.size());
+ plugins = merge(plugins, currentPlugins);
}
}
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/PropertyBasedPluginLocator.java b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/PropertyBasedPluginLocator.java
index d701199..55de7ba 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/PropertyBasedPluginLocator.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/PropertyBasedPluginLocator.java
@@ -18,14 +18,16 @@
*/
package org.apache.apex.engine.plugin.loaders;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.apex.api.plugin.Plugin;
import org.apache.apex.engine.api.plugin.PluginLocator;
+import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import com.datatorrent.stram.StramUtils;
@@ -33,7 +35,7 @@ import com.datatorrent.stram.StramUtils;
/**
* @since 3.6.0
*/
-public class PropertyBasedPluginLocator<T> implements PluginLocator<T>
+public class PropertyBasedPluginLocator<T extends Plugin> implements PluginLocator<T>
{
private static final Logger LOG = LoggerFactory.getLogger(PropertyBasedPluginLocator.class);
private final Class<T> klass;
@@ -46,21 +48,26 @@ public class PropertyBasedPluginLocator<T> implements PluginLocator<T>
}
@Override
- public Collection<T> discoverPlugins(Configuration conf)
+ public Set<T> discoverPlugins(Configuration conf)
{
- List<T> detectedPlugins = new ArrayList<>();
+ Set<T> detectedPlugins = new LinkedHashSet<>();
String classNamesStr = conf.get(this.propertyName);
- if (classNamesStr == null) {
+ if (StringUtils.isBlank(classNamesStr)) {
return detectedPlugins;
}
- String[] classNames = classNamesStr.split(",");
+ Set<String> classNames = new LinkedHashSet<>();
+ Collections.addAll(classNames, classNamesStr.split(","));
for (String className : classNames) {
try {
- Class<? extends T> plugin = StramUtils.classForName(className, this.klass);
- detectedPlugins.add(StramUtils.newInstance(plugin));
+ Class<?> plugin = StramUtils.classForName(className, Object.class);
+ if (klass.isAssignableFrom(plugin)) {
+ detectedPlugins.add(StramUtils.newInstance(plugin.asSubclass(klass)));
+ } else {
+ LOG.info("Skipping loading {} incompatible with {}", className, klass);
+ }
} catch (IllegalArgumentException e) {
- LOG.warn("Could not load plugin {}", className);
+ LOG.warn("Could not load plugin {}", className, e);
}
}
return detectedPlugins;
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ServiceLoaderBasedPluginLocator.java b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ServiceLoaderBasedPluginLocator.java
index 33eba65..e878503 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ServiceLoaderBasedPluginLocator.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/ServiceLoaderBasedPluginLocator.java
@@ -18,18 +18,18 @@
*/
package org.apache.apex.engine.plugin.loaders;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
+import java.util.LinkedHashSet;
import java.util.ServiceLoader;
+import java.util.Set;
+import org.apache.apex.api.plugin.Plugin;
import org.apache.apex.engine.api.plugin.PluginLocator;
import org.apache.hadoop.conf.Configuration;
/**
* @since 3.6.0
*/
-public class ServiceLoaderBasedPluginLocator<T> implements PluginLocator<T>
+public class ServiceLoaderBasedPluginLocator<T extends Plugin> implements PluginLocator<T>
{
private final Class<T> klass;
@@ -39,10 +39,10 @@ public class ServiceLoaderBasedPluginLocator<T> implements PluginLocator<T>
}
@Override
- public Collection<T> discoverPlugins(Configuration conf)
+ public Set<T> discoverPlugins(Configuration conf)
{
- List<T> discovered = new ArrayList<>();
ServiceLoader<T> loader = ServiceLoader.load(this.klass);
+ Set<T> discovered = new LinkedHashSet<>();
for (T plugin : loader) {
discovered.add(plugin);
}
diff --git a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/StaticPluginLocator.java b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/StaticPluginLocator.java
index 2da1665..1809cd4 100644
--- a/engine/src/main/java/org/apache/apex/engine/plugin/loaders/StaticPluginLocator.java
+++ b/engine/src/main/java/org/apache/apex/engine/plugin/loaders/StaticPluginLocator.java
@@ -18,32 +18,35 @@
*/
package org.apache.apex.engine.plugin.loaders;
-import java.util.Arrays;
-import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.apex.api.plugin.Plugin;
import org.apache.apex.engine.api.plugin.PluginLocator;
import org.apache.hadoop.conf.Configuration;
/**
* @since 3.6.0
*/
-public class StaticPluginLocator<T> implements PluginLocator<T>
+public class StaticPluginLocator<T extends Plugin> implements PluginLocator<T>
{
private static final Logger LOG = LoggerFactory.getLogger(StaticPluginLocator.class);
- private final T[] plugins;
+ private final Set<T> plugins;
public StaticPluginLocator(T... plugins)
{
- this.plugins = plugins;
+ this.plugins = new LinkedHashSet<>(plugins.length);
+ Collections.addAll(this.plugins, plugins);
}
@Override
- public Collection<T> discoverPlugins(Configuration conf)
+ public Set<T> discoverPlugins(Configuration conf)
{
- return Arrays.asList(plugins);
+ return plugins;
}
}
--
To stop receiving notification emails like this one, please contact
['"commits@apex.apache.org" <co...@apex.apache.org>'].