You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by br...@apache.org on 2012/10/30 23:59:40 UTC
[2/24] git commit: FLUME-1603: Flume configuration code could be
improved
FLUME-1603: Flume configuration code could be improved
Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/e47fd1e7
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/e47fd1e7
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/e47fd1e7
Branch: refs/heads/FLUME-1502
Commit: e47fd1e720a4faddc2777eb5987d2cc26cb77283
Parents: 831a86f
Author: Brock Noland <br...@apache.org>
Authored: Thu Oct 11 14:49:43 2012 -0500
Committer: Brock Noland <br...@apache.org>
Committed: Tue Oct 30 17:57:23 2012 -0500
----------------------------------------------------------------------
.../org/apache/flume/channel/file/FileChannel.java | 2 +
.../org/apache/flume/channel/jdbc/JdbcChannel.java | 2 +
.../memory/RecoverableMemoryChannel.java | 3 +-
.../org/apache/flume/conf/FlumeConfiguration.java | 10 +-
.../main/java/org/apache/flume/ChannelFactory.java | 3 +-
.../src/main/java/org/apache/flume/Constants.java | 5 +
.../main/java/org/apache/flume/SinkFactory.java | 2 -
.../main/java/org/apache/flume/SourceFactory.java | 1 -
.../flume/channel/DefaultChannelFactory.java | 82 +---
.../org/apache/flume/channel/MemoryChannel.java | 2 +
.../org/apache/flume/sink/DefaultSinkFactory.java | 75 +---
.../apache/flume/source/DefaultSourceFactory.java | 75 +---
.../apache/flume/sink/TestDefaultSinkFactory.java | 23 +-
.../flume/source/TestDefaultSourceFactory.java | 22 +-
flume-ng-node/pom.xml | 6 +
.../file/AbstractFileConfigurationProvider.java | 216 -------
.../flume/conf/file/SimpleNodeConfiguration.java | 75 ---
.../PropertiesFileConfigurationProvider.java | 434 ---------------
.../java/org/apache/flume/node/Application.java | 329 ++++++++----
.../apache/flume/node/ConfigurationProvider.java | 7 +-
.../main/java/org/apache/flume/node/FlumeNode.java | 110 ----
.../org/apache/flume/node/NodeConfiguration.java | 36 --
.../java/org/apache/flume/node/NodeManager.java | 36 --
.../nodemanager/AbstractLogicalNodeManager.java | 69 ---
.../nodemanager/DefaultLogicalNodeManager.java | 263 ---------
.../node/nodemanager/NodeConfigurationAware.java | 37 --
.../TestPropertiesFileConfigurationProvider.java | 82 ---
.../flume/node/TestAbstractLogicalNodeManager.java | 233 --------
.../flume/node/TestDefaultLogicalNodeManager.java | 172 ------
.../java/org/apache/flume/node/TestFlumeNode.java | 111 ----
.../flume/node/TestFlumeNodeApplication.java | 35 --
.../source/FlakeySequenceGeneratorSource.java | 42 --
.../org/apache/flume/source/TestNetcatSource.java | 151 -----
.../src/test/resources/flume-conf.properties | 45 +--
34 files changed, 305 insertions(+), 2491 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
index 6680a2c..a948d28 100644
--- a/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
+++ b/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannel.java
@@ -30,6 +30,7 @@ import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
+import org.apache.flume.annotations.Disposable;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.flume.channel.file.Log.Builder;
@@ -54,6 +55,7 @@ import com.google.common.base.Strings;
* the transaction is synced to disk.
* </p>
*/
+@Disposable
public class FileChannel extends BasicChannelSemantics {
private static final Logger LOG = LoggerFactory
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-channels/flume-jdbc-channel/src/main/java/org/apache/flume/channel/jdbc/JdbcChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-jdbc-channel/src/main/java/org/apache/flume/channel/jdbc/JdbcChannel.java b/flume-ng-channels/flume-jdbc-channel/src/main/java/org/apache/flume/channel/jdbc/JdbcChannel.java
index bca0c50..1c64040 100644
--- a/flume-ng-channels/flume-jdbc-channel/src/main/java/org/apache/flume/channel/jdbc/JdbcChannel.java
+++ b/flume-ng-channels/flume-jdbc-channel/src/main/java/org/apache/flume/channel/jdbc/JdbcChannel.java
@@ -21,12 +21,14 @@ import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.annotations.Disposable;
import org.apache.flume.channel.AbstractChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* <p>A JDBC based channel implementation.</p>
*/
+@Disposable
public class JdbcChannel extends AbstractChannel {
private static final Logger LOG = LoggerFactory.getLogger(JdbcChannel.class);
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-channels/flume-recoverable-memory-channel/src/main/java/org/apache/flume/channel/recoverable/memory/RecoverableMemoryChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-recoverable-memory-channel/src/main/java/org/apache/flume/channel/recoverable/memory/RecoverableMemoryChannel.java b/flume-ng-channels/flume-recoverable-memory-channel/src/main/java/org/apache/flume/channel/recoverable/memory/RecoverableMemoryChannel.java
index 49e7cfd..d2201b9 100644
--- a/flume-ng-channels/flume-recoverable-memory-channel/src/main/java/org/apache/flume/channel/recoverable/memory/RecoverableMemoryChannel.java
+++ b/flume-ng-channels/flume-recoverable-memory-channel/src/main/java/org/apache/flume/channel/recoverable/memory/RecoverableMemoryChannel.java
@@ -31,6 +31,7 @@ import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
+import org.apache.flume.annotations.Disposable;
import org.apache.flume.channel.BasicChannelSemantics;
import org.apache.flume.channel.BasicTransactionSemantics;
import org.apache.flume.channel.MemoryChannel;
@@ -54,7 +55,7 @@ import com.google.common.collect.Lists;
* {@link org.apache.flume.channel.file.FileChannel}, which gives better
* performance and is also durable.
*/
-
+@Disposable
@Deprecated
public class RecoverableMemoryChannel extends BasicChannelSemantics {
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
----------------------------------------------------------------------
diff --git a/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java b/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
index 9b209e8..b81c2a5 100644
--- a/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
+++ b/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
@@ -19,14 +19,12 @@ package org.apache.flume.conf;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
@@ -70,14 +68,12 @@ public class FlumeConfiguration {
/**
* Creates a populated Flume Configuration object.
*/
- public FlumeConfiguration(Properties properties) {
+ public FlumeConfiguration(Map<String, String> properties) {
agentConfigMap = new HashMap<String, AgentConfiguration>();
errors = new LinkedList<FlumeConfigurationError>();
// Construct the in-memory component hierarchy
- Enumeration<?> propertyNames = properties.propertyNames();
- while (propertyNames.hasMoreElements()) {
- String name = (String) propertyNames.nextElement();
- String value = properties.getProperty(name);
+ for(String name : properties.keySet()) {
+ String value = properties.get(name);
if (!addRawProperty(name, value)) {
logger.warn("Configuration property ignored: " + name + " = " + value);
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java b/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
index e269909..8672fda 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
@@ -22,6 +22,7 @@ public interface ChannelFactory {
public Channel create(String name, String type) throws FlumeException;
- public boolean unregister(Channel channel);
+ public Class<? extends Channel> getChannelClass(String type)
+ throws FlumeException;
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/Constants.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/Constants.java b/flume-ng-core/src/main/java/org/apache/flume/Constants.java
index 4c6992d..c5f46ed 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/Constants.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/Constants.java
@@ -20,6 +20,11 @@ package org.apache.flume;
public final class Constants {
+ /**
+ * Disables the fail-fast startup behavior. This would be used in the
+ * scenario where the agent is expected to start, but the config
+ * file be populated at a later point in time.
+ */
public static final String SYSPROP_CALLED_FROM_SERVICE
= "flume.called.from.service";
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/SinkFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/SinkFactory.java b/flume-ng-core/src/main/java/org/apache/flume/SinkFactory.java
index d8fd0da..369b11c 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/SinkFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/SinkFactory.java
@@ -25,6 +25,4 @@ public interface SinkFactory {
public Sink create(String name, String type)
throws FlumeException;
- public boolean unregister(Sink sink);
-
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/SourceFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/SourceFactory.java b/flume-ng-core/src/main/java/org/apache/flume/SourceFactory.java
index 91cc866..38b6e82 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/SourceFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/SourceFactory.java
@@ -25,5 +25,4 @@ public interface SourceFactory {
public Source create(String sourceName, String type)
throws FlumeException;
- public boolean unregister(Source source);
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java b/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
index ee32696..05ecf8b 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
@@ -19,9 +19,6 @@
package org.apache.flume.channel;
-import java.util.HashMap;
-import java.util.Map;
-
import org.apache.flume.Channel;
import org.apache.flume.ChannelFactory;
import org.apache.flume.FlumeException;
@@ -36,40 +33,30 @@ public class DefaultChannelFactory implements ChannelFactory {
private static final Logger logger = LoggerFactory
.getLogger(DefaultChannelFactory.class);
- private Map<Class<?>, Map<String, Channel>> channels;
-
- public DefaultChannelFactory() {
- channels = new HashMap<Class<?>, Map<String, Channel>>();
- }
-
@Override
- public boolean unregister(Channel channel) {
- Preconditions.checkNotNull(channel);
- logger.info("Unregister channel {}", channel);
- boolean removed = false;
+ public Channel create(String name, String type) throws FlumeException {
+ Preconditions.checkNotNull(name, "name");
+ Preconditions.checkNotNull(type, "type");
+ logger.debug("Creating instance of channel {} type {}", name, type);
- Map<String, Channel> channelMap = channels.get(channel.getClass());
- if (channelMap != null) {
- removed = (channelMap.remove(channel.getName()) != null);
- if (channelMap.size() == 0) {
- channels.remove(channel.getClass());
- }
+ Class<? extends Channel> channelClass = getChannelClass(type);
+ try {
+ return channelClass.newInstance();
+ } catch (Exception ex) {
+ // Clean up channel map
+ throw new FlumeException("Unable to create channel: " + name
+ + ", type: " + type + ", class: " + channelClass, ex);
}
-
- return removed;
}
@SuppressWarnings("unchecked")
- @Override
- public Channel create(String name, String type) throws FlumeException {
- Preconditions.checkNotNull(name);
- Preconditions.checkNotNull(type);
- logger.debug("Creating instance of channel {} type {}", name, type);
+ public Class<? extends Channel> getChannelClass(String type)
+ throws FlumeException {
String channelClassName = type;
-
ChannelType channelType = ChannelType.OTHER;
+
try {
channelType = ChannelType.valueOf(type.toUpperCase());
} catch (IllegalArgumentException ex) {
@@ -79,50 +66,11 @@ public class DefaultChannelFactory implements ChannelFactory {
if (!channelType.equals(ChannelType.OTHER)) {
channelClassName = channelType.getChannelClassName();
}
-
- Class<? extends Channel> channelClass = null;
try {
- channelClass = (Class<? extends Channel>) Class.forName(channelClassName);
+ return (Class<? extends Channel>) Class.forName(channelClassName);
} catch (Exception ex) {
throw new FlumeException("Unable to load channel type: " + type
+ ", class: " + channelClassName, ex);
}
-
- Map<String, Channel> channelMap = channels.get(channelClass);
- if (channelMap == null) {
- channelMap = new HashMap<String, Channel>();
- channels.put(channelClass, channelMap);
- }
-
- Channel channel = channelMap.get(name);
-
- if (channel == null) {
- try {
- channel = channelClass.newInstance();
- channel.setName(name);
- channelMap.put(name, channel);
- } catch (Exception ex) {
- // Clean up channel map
- channels.remove(channelClass);
- throw new FlumeException("Unable to create channel: " + name
- + ", type: " + type + ", class: " + channelClassName, ex);
- }
- }
-
- return channel;
- }
-
- public Map<Class<?>, Map<String, Channel>> getRegistryClone() {
- Map<Class<?>, Map<String, Channel>> result =
- new HashMap<Class<?>, Map<String, Channel>>();
-
- for (Class<?> klass : channels.keySet()) {
- Map<String, Channel> channelMap = channels.get(klass);
- Map<String, Channel> resultChannelMap = new HashMap<String, Channel>();
- resultChannelMap.putAll(channelMap);
- result.put(klass, resultChannelMap);
- }
-
- return result;
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
index 06c90d9..d1e542f 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
@@ -27,12 +27,14 @@ import javax.annotation.concurrent.GuardedBy;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
+import org.apache.flume.annotations.Recyclable;
import org.apache.flume.instrumentation.ChannelCounter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+@Recyclable
public class MemoryChannel extends BasicChannelSemantics {
private static Logger LOGGER = LoggerFactory.getLogger(MemoryChannel.class);
private static final Integer defaultCapacity = 100;
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkFactory.java b/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkFactory.java
index e71d44e..447efb3 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/sink/DefaultSinkFactory.java
@@ -19,9 +19,6 @@
package org.apache.flume.sink;
-import java.util.HashMap;
-import java.util.Map;
-
import org.apache.flume.FlumeException;
import org.apache.flume.Sink;
import org.apache.flume.SinkFactory;
@@ -36,37 +33,12 @@ public class DefaultSinkFactory implements SinkFactory {
private static final Logger logger = LoggerFactory
.getLogger(DefaultSinkFactory.class);
- private final Map<Class<?>, Map<String, Sink>> sinks;
-
- public DefaultSinkFactory() {
- sinks = new HashMap<Class<?>, Map<String, Sink>>();
- }
-
- @Override
- public synchronized boolean unregister(Sink sink) {
- Preconditions.checkNotNull(sink);
- boolean removed = false;
-
- logger.debug("Unregistering sink {}", sink);
-
- Map<String, Sink> sinkMap = sinks.get(sink.getClass());
- if (sinkMap != null) {
- removed = (sinkMap.remove(sink.getName()) != null);
-
- if (sinkMap.size() == 0) {
- sinks.remove(sink.getClass());
- }
- }
-
- return removed;
- }
-
@SuppressWarnings("unchecked")
@Override
public Sink create(String name, String type)
throws FlumeException {
- Preconditions.checkNotNull(name);
- Preconditions.checkNotNull(type);
+ Preconditions.checkNotNull(name, "name");
+ Preconditions.checkNotNull(type, "type");
logger.info("Creating instance of sink: {}, type: {}", name, type);
String sinkClassName = type;
@@ -89,42 +61,13 @@ public class DefaultSinkFactory implements SinkFactory {
throw new FlumeException("Unable to load sink type: " + type
+ ", class: " + sinkClassName, ex);
}
-
- Map<String, Sink> sinkMap = sinks.get(sinkClass);
- if (sinkMap == null) {
- sinkMap = new HashMap<String, Sink>();
- sinks.put(sinkClass, sinkMap);
- }
-
- Sink sink = sinkMap.get(name);
-
- if (sink == null) {
- try {
- sink = sinkClass.newInstance();
- sink.setName(name);
- sinkMap.put(name, sink);
- } catch (Exception ex) {
- // Clean up the sink map
- sinks.remove(sinkClass);
- throw new FlumeException("Unable to create sink: " + name
- + ", type: " + type + ", class: " + sinkClassName, ex);
- }
- }
-
- return sink;
- }
-
- public synchronized Map<Class<?>, Map<String, Sink>> getRegistryClone() {
- Map<Class<?>, Map<String, Sink>> result =
- new HashMap<Class<?>, Map<String, Sink>>();
-
- for (Class<?> klass : sinks.keySet()) {
- Map<String, Sink> sinkMap = sinks.get(klass);
- Map<String, Sink> resultSinkMap = new HashMap<String, Sink>();
- resultSinkMap.putAll(sinkMap);
- result.put(klass, resultSinkMap);
+ try {
+ Sink sink = sinkClass.newInstance();
+ sink.setName(name);
+ return sink;
+ } catch (Exception ex) {
+ throw new FlumeException("Unable to create sink: " + name
+ + ", type: " + type + ", class: " + sinkClassName, ex);
}
-
- return result;
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/main/java/org/apache/flume/source/DefaultSourceFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/DefaultSourceFactory.java b/flume-ng-core/src/main/java/org/apache/flume/source/DefaultSourceFactory.java
index 18533ae..02ca810 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/DefaultSourceFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/DefaultSourceFactory.java
@@ -18,9 +18,6 @@
*/
package org.apache.flume.source;
-import java.util.HashMap;
-import java.util.Map;
-
import org.apache.flume.FlumeException;
import org.apache.flume.Source;
import org.apache.flume.SourceFactory;
@@ -35,35 +32,6 @@ public class DefaultSourceFactory implements SourceFactory {
private static final Logger logger = LoggerFactory
.getLogger(DefaultSourceFactory.class);
- /**
- * Cache of sources created thus far. The outer map is keyed on the source
- * type and the inner map is keyed on source name.
- */
- private final Map<Class<?>, Map<String, Source>> sources;
-
- public DefaultSourceFactory() {
- sources = new HashMap<Class<?>, Map<String, Source>>();
- }
-
- @Override
- public synchronized boolean unregister(Source source) {
- Preconditions.checkNotNull(source);
- boolean removed = false;
-
- logger.debug("Unregistering source {}", source);
-
- Map<String, Source> sourceMap = sources.get(source.getClass());
- if (sourceMap != null) {
- removed = (sourceMap.remove(source.getName()) != null);
-
- if (sourceMap.size() == 0) {
- sources.remove(source.getClass());
- }
- }
-
- return removed;
- }
-
@SuppressWarnings("unchecked")
@Override
public synchronized Source create(String name, String type)
@@ -93,42 +61,13 @@ public class DefaultSourceFactory implements SourceFactory {
throw new FlumeException("Unable to load source type: " + type
+ ", class: " + sourceClassName, ex);
}
-
- Map<String, Source> sourceMap = sources.get(sourceClass);
- if (sourceMap == null) {
- sourceMap = new HashMap<String, Source>();
- sources.put(sourceClass, sourceMap);
- }
-
- Source source = sourceMap.get(name);
-
- if (source == null) {
- try {
- source = sourceClass.newInstance();
- source.setName(name);
- sourceMap.put(name, source);
- } catch (Exception ex) {
- // Clean up the source map
- sources.remove(sourceClass);
- throw new FlumeException("Unable to create source: " + name
- +", type: " + type + ", class: " + sourceClassName, ex);
- }
- }
-
- return source;
- }
-
- public synchronized Map<Class<?>, Map<String, Source>> getRegistryClone() {
- Map<Class<?>, Map<String, Source>> result =
- new HashMap<Class<?>, Map<String, Source>>();
-
- for (Class<?> klass : sources.keySet()) {
- Map<String, Source> srcMap = sources.get(klass);
- Map<String, Source> resultSrcMap = new HashMap<String, Source>();
- resultSrcMap.putAll(srcMap);
- result.put(klass, resultSrcMap);
+ try {
+ Source source = sourceClass.newInstance();
+ source.setName(name);
+ return source;
+ } catch (Exception ex) {
+ throw new FlumeException("Unable to create source: " + name
+ +", type: " + type + ", class: " + sourceClassName, ex);
}
-
- return result;
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
index ab3f447..835f541 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
@@ -19,8 +19,6 @@
package org.apache.flume.sink;
-import java.util.Map;
-
import org.apache.flume.Sink;
import org.apache.flume.SinkFactory;
import org.junit.Assert;
@@ -52,8 +50,8 @@ public class TestDefaultSinkFactory {
Sink s1 = sinkFactory.create("avroSink1", "avro");
Sink s2 = sinkFactory.create("avroSink2", "avro");
- Assert.assertSame(avroSink1, s1);
- Assert.assertSame(avroSink2, s2);
+ Assert.assertNotSame(avroSink1, s1);
+ Assert.assertNotSame(avroSink2, s2);
}
private void verifySinkCreation(String name, String type, Class<?> typeClass)
@@ -71,21 +69,4 @@ public class TestDefaultSinkFactory {
verifySinkCreation("avro-sink", "avro", AvroSink.class);
}
-
- @Test
- public void testSinkRegistry() {
- Sink s1 = sinkFactory.create("s1", "avro");
- Map<Class<?>, Map<String, Sink>> sr =
- ((DefaultSinkFactory) sinkFactory).getRegistryClone();
-
- Assert.assertEquals(1, sr.size());
- Map<String, Sink> sinkMap = sr.get(AvroSink.class);
- Assert.assertNotNull(sinkMap);
- Assert.assertEquals(1, sinkMap.size());
-
- Sink sink = sinkMap.get("s1");
- Assert.assertNotNull(sink);
- Assert.assertSame(s1, sink);
- }
-
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-core/src/test/java/org/apache/flume/source/TestDefaultSourceFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestDefaultSourceFactory.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestDefaultSourceFactory.java
index ff5b4d6..ffbe177 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestDefaultSourceFactory.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestDefaultSourceFactory.java
@@ -19,8 +19,6 @@
package org.apache.flume.source;
-import java.util.Map;
-
import org.apache.flume.Source;
import org.apache.flume.SourceFactory;
import org.junit.Assert;
@@ -51,8 +49,8 @@ public class TestDefaultSourceFactory {
Source s1 = sourceFactory.create("avroSource1", "avro");
Source s2 = sourceFactory.create("avroSource2", "avro");
- Assert.assertSame(avroSource1, s1);
- Assert.assertSame(avroSource2, s2);
+ Assert.assertNotSame(avroSource1, s1);
+ Assert.assertNotSame(avroSource2, s2);
}
@@ -73,20 +71,4 @@ public class TestDefaultSourceFactory {
MockSource.class);
}
- @Test
- public void testSourceRegistry() throws Exception {
- Source s1 = sourceFactory.create("s1", "avro");
- Map<Class<?>, Map<String, Source>> sr =
- ((DefaultSourceFactory) sourceFactory).getRegistryClone();
-
- Assert.assertEquals(1, sr.size());
-
- Map<String, Source> srMap = sr.get(AvroSource.class);
- Assert.assertNotNull(srMap);
- Assert.assertEquals(1, srMap.size());
-
- Source src = srMap.get("s1");
- Assert.assertNotNull(src);
- Assert.assertSame(s1, src);
- }
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/pom.xml
----------------------------------------------------------------------
diff --git a/flume-ng-node/pom.xml b/flume-ng-node/pom.xml
index 6a0fe15..035ae06 100644
--- a/flume-ng-node/pom.xml
+++ b/flume-ng-node/pom.xml
@@ -125,6 +125,12 @@
</dependency>
<dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-core-asl</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/conf/file/AbstractFileConfigurationProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/conf/file/AbstractFileConfigurationProvider.java b/flume-ng-node/src/main/java/org/apache/flume/conf/file/AbstractFileConfigurationProvider.java
deleted file mode 100644
index a2c882b..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/conf/file/AbstractFileConfigurationProvider.java
+++ /dev/null
@@ -1,216 +0,0 @@
-/**
- * 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.flume.conf.file;
-
-import java.io.File;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.flume.ChannelFactory;
-import org.apache.flume.CounterGroup;
-import org.apache.flume.SinkFactory;
-import org.apache.flume.SourceFactory;
-import org.apache.flume.lifecycle.LifecycleState;
-import org.apache.flume.node.ConfigurationProvider;
-import org.apache.flume.node.nodemanager.NodeConfigurationAware;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-public abstract class AbstractFileConfigurationProvider implements
- ConfigurationProvider {
-
- private final Logger logger = LoggerFactory.getLogger(getClass());
-
- private File file;
- private ChannelFactory channelFactory;
- private SourceFactory sourceFactory;
- private SinkFactory sinkFactory;
- private String nodeName;
- private NodeConfigurationAware configurationAware;
-
- private LifecycleState lifecycleState;
- private ScheduledExecutorService executorService;
- private CounterGroup counterGroup;
-
- public AbstractFileConfigurationProvider() {
- lifecycleState = LifecycleState.IDLE;
- counterGroup = new CounterGroup();
- }
-
- @Override
- public String toString() {
- return "{ file:" + file + " counterGroup:" + counterGroup + " provider:"
- + getClass().getCanonicalName() + " nodeName:" + nodeName + " }";
- }
-
- @Override
- public void start() {
- logger.info("Configuration provider starting");
-
- Preconditions.checkState(file != null,
- "The parameter file must not be null");
-
- executorService = Executors.newSingleThreadScheduledExecutor(
- new ThreadFactoryBuilder().setNameFormat("conf-file-poller-%d")
- .build());
-
- FileWatcherRunnable fileWatcherRunnable = new FileWatcherRunnable();
-
- fileWatcherRunnable.file = file;
- fileWatcherRunnable.counterGroup = counterGroup;
-
- executorService.scheduleWithFixedDelay(fileWatcherRunnable, 0, 30,
- TimeUnit.SECONDS);
-
- lifecycleState = LifecycleState.START;
-
- logger.debug("Configuration provider started");
- }
-
- @Override
- public void stop() {
- logger.info("Configuration provider stopping");
-
- executorService.shutdown();
-
- while (!executorService.isTerminated()) {
- try {
- logger.debug("Waiting for file watcher to terminate");
- executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- logger.debug("Interrupted while waiting for file watcher to terminate");
- Thread.currentThread().interrupt();
- }
- }
-
- lifecycleState = LifecycleState.STOP;
-
- logger.debug("Configuration provider stopped");
- }
-
- protected abstract void load();
-
- // Synchronized wrapper to call the load function
- private synchronized void doLoad() {
- Preconditions
- .checkState(nodeName != null,
- "No node name specified - Unable to determine what part of the config to load");
- Preconditions.checkState(channelFactory != null,
- "No channel factory configured");
- Preconditions.checkState(sourceFactory != null,
- "No source factory configured");
- Preconditions.checkState(sinkFactory != null, "No sink factory configured");
-
- load();
- }
-
- public File getFile() {
- return file;
- }
-
- public void setFile(File file) {
- this.file = file;
- }
-
- @Override
- public LifecycleState getLifecycleState() {
- return lifecycleState;
- }
-
- public ChannelFactory getChannelFactory() {
- return channelFactory;
- }
-
- public void setChannelFactory(ChannelFactory channelFactory) {
- this.channelFactory = channelFactory;
- }
-
- public SourceFactory getSourceFactory() {
- return sourceFactory;
- }
-
- public void setSourceFactory(SourceFactory sourceFactory) {
- this.sourceFactory = sourceFactory;
- }
-
- public SinkFactory getSinkFactory() {
- return sinkFactory;
- }
-
- public void setSinkFactory(SinkFactory sinkFactory) {
- this.sinkFactory = sinkFactory;
- }
-
- public NodeConfigurationAware getConfigurationAware() {
- return configurationAware;
- }
-
- public void setConfigurationAware(NodeConfigurationAware configurationAware) {
- this.configurationAware = configurationAware;
- }
-
- public String getNodeName() {
- return nodeName;
- }
-
- public void setNodeName(String nodeName) {
- this.nodeName = nodeName;
- }
-
- public class FileWatcherRunnable implements Runnable {
-
- private File file;
- private CounterGroup counterGroup;
-
- private long lastChange;
-
- @Override
- public void run() {
- logger.debug("Checking file:{} for changes", file);
-
- counterGroup.incrementAndGet("file.checks");
-
- long lastModified = file.lastModified();
-
- if (lastModified > lastChange) {
- logger.info("Reloading configuration file:{}", file);
-
- counterGroup.incrementAndGet("file.loads");
-
- lastChange = lastModified;
-
- try {
- doLoad();
- } catch (Exception e) {
- logger.error("Failed to load configuration data. Exception follows.",
- e);
- } catch (NoClassDefFoundError e) {
- logger.error("Failed to start agent because dependencies were not " +
- "found in classpath. Error follows.", e);
- } catch (Throwable t) {
- // caught because the caller does not handle or log Throwables
- logger.error("Unhandled error", t);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/conf/file/SimpleNodeConfiguration.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/conf/file/SimpleNodeConfiguration.java b/flume-ng-node/src/main/java/org/apache/flume/conf/file/SimpleNodeConfiguration.java
deleted file mode 100644
index 99b8bcc..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/conf/file/SimpleNodeConfiguration.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.flume.conf.file;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.flume.Channel;
-import org.apache.flume.SinkRunner;
-import org.apache.flume.SourceRunner;
-import org.apache.flume.node.NodeConfiguration;
-
-public class SimpleNodeConfiguration implements NodeConfiguration {
-
- private Map<String, Channel> channels;
- private Map<String, SourceRunner> sourceRunners;
- private Map<String, SinkRunner> sinkRunners;
-
- public SimpleNodeConfiguration() {
- channels = new HashMap<String, Channel>();
- sourceRunners = new HashMap<String, SourceRunner>();
- sinkRunners = new HashMap<String, SinkRunner>();
- }
-
- @Override
- public String toString() {
- return "{ sourceRunners:" + sourceRunners + " sinkRunners:" + sinkRunners
- + " channels:" + channels + " }";
- }
-
- @Override
- public Map<String, Channel> getChannels() {
- return channels;
- }
-
- public void setChannels(Map<String, Channel> channels) {
- this.channels = channels;
- }
-
- @Override
- public Map<String, SourceRunner> getSourceRunners() {
- return sourceRunners;
- }
-
- public void setSourceRunners(Map<String, SourceRunner> sourceRunners) {
- this.sourceRunners = sourceRunners;
- }
-
- @Override
- public Map<String, SinkRunner> getSinkRunners() {
- return sinkRunners;
- }
-
- public void setSinkRunners(Map<String, SinkRunner> sinkRunners) {
- this.sinkRunners = sinkRunners;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java b/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
deleted file mode 100644
index 8dbbe57..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
+++ /dev/null
@@ -1,434 +0,0 @@
-/**
- * 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.flume.conf.properties;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Set;
-
-import org.apache.flume.Channel;
-import org.apache.flume.ChannelSelector;
-import org.apache.flume.Context;
-import org.apache.flume.Sink;
-import org.apache.flume.SinkProcessor;
-import org.apache.flume.SinkRunner;
-import org.apache.flume.Source;
-import org.apache.flume.SourceRunner;
-import org.apache.flume.channel.ChannelProcessor;
-import org.apache.flume.channel.ChannelSelectorFactory;
-import org.apache.flume.conf.BasicConfigurationConstants;
-import org.apache.flume.conf.ComponentConfiguration;
-import org.apache.flume.conf.Configurables;
-import org.apache.flume.conf.FlumeConfiguration;
-import org.apache.flume.conf.FlumeConfiguration.AgentConfiguration;
-import org.apache.flume.conf.channel.ChannelSelectorConfiguration;
-import org.apache.flume.conf.file.AbstractFileConfigurationProvider;
-import org.apache.flume.conf.file.SimpleNodeConfiguration;
-import org.apache.flume.conf.sink.SinkConfiguration;
-import org.apache.flume.conf.sink.SinkGroupConfiguration;
-import org.apache.flume.conf.source.SourceConfiguration;
-import org.apache.flume.node.NodeConfiguration;
-import org.apache.flume.sink.DefaultSinkProcessor;
-import org.apache.flume.sink.SinkGroup;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * <p>
- * A configuration provider that uses properties file for specifying
- * configuration. The configuration files follow the Java properties file syntax
- * rules specified at {@link java.util.Properties#load(java.io.Reader)}. Every
- * configuration value specified in the properties file is prefixed by an
- * <em>Agent Name</em> which helps isolate an individual agent's namespace.
- * </p>
- * <p>
- * Valid configuration files must observe the following rules for every agent
- * namespace.
- * <ul>
- * <li>For every <agent name> there must be three lists specified that
- * include <tt><agent name>.sources</tt>,
- * <tt><agent name>.sinks</tt>, and <tt><agent name>.channels</tt>.
- * Each of these lists must contain a space separated list of names
- * corresponding to that particular entity.</li>
- * <li>For each source named in <tt><agent name>.sources</tt>, there must
- * be a non-empty <tt>type</tt> attribute specified from the valid set of source
- * types. For example:
- * <tt><agent name>.sources.<source name>.type = event</tt></li>
- * <li>For each source named in <tt><agent name>.sources</tt>, there must
- * be a space-separated list of channel names that the source will associate
- * with during runtime. Each of these names must be contained in the channels
- * list specified by <tt><agent name>.channels</tt>. For example:
- * <tt><agent name>.sources.<source name>.channels =
- * <channel-1 name> <channel-2 name></tt></li>
- * <li>For each source named in the <tt><agent name>.sources</tt>, there
- * must be a <tt>runner</tt> namespace of configuration that configures the
- * associated source runner. For example:
- * <tt><agent name>.sources.<source name>.runner.type = avro</tt>.
- * This namespace can also be used to configure other configuration of the
- * source runner as needed. For example:
- * <tt><agent name>.sources.<source name>.runner.port = 10101</tt>
- * </li>
- * <li>For each source named in <tt><sources>.sources</tt> there can
- * be an optional <tt>selector.type</tt> specified that identifies the type
- * of channel selector associated with the source. If not specified, the
- * default replicating channel selector is used.
- * </li><li>For each channel named in the <tt><agent name>.channels</tt>,
- * there must be a non-empty <tt>type</tt> attribute specified from the valid
- * set of channel types. For example:
- * <tt><agent name>.channels.<channel name>.type = mem</tt></li>
- * <li>For each sink named in the <tt><agent name>.sinks</tt>, there must
- * be a non-empty <tt>type</tt> attribute specified from the valid set of sink
- * types. For example:
- * <tt><agent name>.sinks.<sink name>.type = hdfs</tt></li>
- * <li>For each sink named in the <tt><agent name>.sinks</tt>, there must
- * be a non-empty single-valued channel name specified as the value of the
- * <tt>channel</tt> attribute. This value must be contained in the channels list
- * specified by <tt><agent name>.channels</tt>. For example:
- * <tt><agent name>.sinks.<sink name>.channel =
- * <channel name></tt></li>
- * <li>For each sink named in the <tt><agent name>.sinks</tt>, there must
- * be a <tt>runner</tt> namespace of configuration that configures the
- * associated sink runner. For example:
- * <tt><agent name>.sinks.<sink name>.runner.type = polling</tt>.
- * This namespace can also be used to configure other configuration of the sink
- * runner as needed. For example:
- * <tt><agent name>.sinks.<sink name>.runner.polling.interval =
- * 60</tt></li>
- * <li>A fourth optional list <tt><agent name>.sinkgroups</tt>
- * may be added to each agent, consisting of unique space separated names
- * for groups</li>
- * <li>Each sinkgroup must specify sinks, containing a list of all sinks
- * belonging to it. These cannot be shared by multiple groups.
- * Further, one can set a processor and behavioral parameters to determine
- * how sink selection is made via <tt><agent name>.sinkgroups.<
- * group name<.processor</tt>. For further detail refer to inividual processor
- * documentation</li>
- * <li>Sinks not assigned to a group will be assigned to default single sink
- * groups.</li>
- * </ul>
- *
- * Apart from the above required configuration values, each source, sink or
- * channel can have its own set of arbitrary configuration as required by the
- * implementation. Each of these configuration values are expressed by fully
- * namespace qualified configuration keys. For example, the configuration
- * property called <tt>capacity</tt> for a channel called <tt>ch1</tt> for the
- * agent named <tt>host1</tt> with value <tt>1000</tt> will be expressed as:
- * <tt>host1.channels.ch1.capacity = 1000</tt>.
- * </p>
- * <p>
- * Any information contained in the configuration file other than what pertains
- * to the configured agents, sources, sinks and channels via the explicitly
- * enumerated list of sources, sinks and channels per agent name are ignored by
- * this provider. Moreover, if any of the required configuration values are not
- * present in the configuration file for the configured entities, that entity
- * and anything that depends upon it is considered invalid and consequently not
- * configured. For example, if a channel is missing its <tt>type</tt> attribute,
- * it is considered misconfigured. Also, any sources or sinks that depend upon
- * this channel are also considered misconfigured and not initialized.
- * </p>
- * <p>
- * Example configuration file:
- *
- * <pre>
- * #
- * # Flume Configuration
- * # This file contains configuration for one Agent identified as host1.
- * #
- *
- * host1.sources = avroSource thriftSource
- * host1.channels = jdbcChannel
- * host1.sinks = hdfsSink
- *
- * # avroSource configuration
- * host1.sources.avroSource.type = org.apache.flume.source.AvroSource
- * host1.sources.avroSource.runner.type = avro
- * host1.sources.avroSource.runner.port = 11001
- * host1.sources.avroSource.channels = jdbcChannel
- * host1.sources.avroSource.selector.type = replicating
- *
- * # thriftSource configuration
- * host1.sources.thriftSource.type = org.apache.flume.source.ThriftSource
- * host1.sources.thriftSource.runner.type = thrift
- * host1.sources.thriftSource.runner.port = 12001
- * host1.sources.thriftSource.channels = jdbcChannel
- *
- * # jdbcChannel configuration
- * host1.channels.jdbcChannel.type = jdbc
- * host1.channels.jdbcChannel.jdbc.driver = com.mysql.jdbc.Driver
- * host1.channels.jdbcChannel.jdbc.connect.url = http://localhost/flumedb
- * host1.channels.jdbcChannel.jdbc.username = flume
- * host1.channels.jdbcChannel.jdbc.password = flume
- *
- * # hdfsSink configuration
- * host1.sinks.hdfsSink.type = hdfs
- * host1.sinks.hdfsSink.hdfs.path = hdfs://localhost/
- * host1.sinks.hdfsSink.batchsize = 1000
- * host1.sinks.hdfsSink.runner.type = polling
- * host1.sinks.hdfsSink.runner.polling.interval = 60
- * </pre>
- *
- * </p>
- *
- * @see java.util.Properties#load(java.io.Reader)
- */
-public class PropertiesFileConfigurationProvider extends
- AbstractFileConfigurationProvider {
-
- private static final Logger LOGGER = LoggerFactory
- .getLogger(PropertiesFileConfigurationProvider.class);
-
- @Override
- protected void load() {
- File propertiesFile = getFile();
- BufferedReader reader = null;
-
- try {
- reader = new BufferedReader(new FileReader(propertiesFile));
- Properties properties = new Properties();
- properties.load(reader);
- // If there is an old configuration, shut it down.
-
- NodeConfiguration conf = new SimpleNodeConfiguration();
- FlumeConfiguration fconfig = new FlumeConfiguration(properties);
- AgentConfiguration agentConf = fconfig.getConfigurationFor(getNodeName());
-
- if (agentConf != null) {
- getConfigurationAware().stopAllComponents();
-
- loadChannels(agentConf, conf);
- loadSources(agentConf, conf);
- loadSinks(agentConf, conf);
-
- getConfigurationAware().startAllComponents(conf);
- } else {
- LOGGER.warn("No configuration found for this host:{}", getNodeName());
- }
- } catch (IOException ex) {
- LOGGER.error("Unable to load file:" + propertiesFile
- + " (I/O failure) - Exception follows.", ex);
- } catch (InstantiationException ex) {
- LOGGER.error("Unable to load file:" + propertiesFile
- + " (failed to instantiate component) - Exception follows.", ex);
- } finally {
- if (reader != null) {
- try {
- reader.close();
- } catch (IOException ex) {
- LOGGER.warn(
- "Unable to close file reader for file: " + propertiesFile, ex);
- }
- }
- }
- }
-
- private void loadChannels(AgentConfiguration agentConf,
- NodeConfiguration conf) throws InstantiationException {
- LOGGER.info("Creating channels");
- Set<String> channels = agentConf.getChannelSet();
- Map<String, ComponentConfiguration> compMap =
- agentConf.getChannelConfigMap();
- for (String chName : channels) {
- ComponentConfiguration comp = compMap.get(chName);
- if(comp != null) {
- Channel channel = getChannelFactory().create(comp.getComponentName(),
- comp.getType());
-
- Configurables.configure(channel, comp);
-
- conf.getChannels().put(comp.getComponentName(), channel);
- }
- }
-
- for (String ch : channels) {
- Context context = agentConf.getChannelContext().get(ch);
- if(context != null){
- Channel channel =
- getChannelFactory().create(ch, context.getString(
- BasicConfigurationConstants.CONFIG_TYPE));
- Configurables.configure(channel, context);
- conf.getChannels().put(ch, channel);
- LOGGER.info("created channel " + ch);
- }
- }
-
- }
-
- private void loadSources(AgentConfiguration agentConf, NodeConfiguration conf)
- throws InstantiationException {
-
- Set<String> sources = agentConf.getSourceSet();
- Map<String, ComponentConfiguration> compMap =
- agentConf.getSourceConfigMap();
- for (String sourceName : sources) {
- ComponentConfiguration comp = compMap.get(sourceName);
- if(comp != null) {
- SourceConfiguration config = (SourceConfiguration) comp;
-
- Source source = getSourceFactory().create(comp.getComponentName(),
- comp.getType());
-
- Configurables.configure(source, config);
- Set<String> channelNames = config.getChannels();
- List<Channel> channels = new ArrayList<Channel>();
- for (String chName : channelNames) {
- channels.add(conf.getChannels().get(chName));
- }
-
- ChannelSelectorConfiguration selectorConfig =
- config.getSelectorConfiguration();
-
- ChannelSelector selector = ChannelSelectorFactory.create(
- channels, selectorConfig);
-
- ChannelProcessor channelProcessor = new ChannelProcessor(selector);
- Configurables.configure(channelProcessor, config);
-
- source.setChannelProcessor(channelProcessor);
- conf.getSourceRunners().put(comp.getComponentName(),
- SourceRunner.forSource(source));
- }
- }
- Map<String, Context> sourceContexts = agentConf.getSourceContext();
-
- for (String src : sources) {
- Context context = sourceContexts.get(src);
- if(context != null){
- Source source =
- getSourceFactory().create(src,
- context.getString(BasicConfigurationConstants.CONFIG_TYPE));
- List<Channel> channels = new ArrayList<Channel>();
- Configurables.configure(source, context);
- String[] channelNames = context.getString(
- BasicConfigurationConstants.CONFIG_CHANNELS).split("\\s+");
- for (String chName : channelNames) {
- channels.add(conf.getChannels().get(chName));
- }
-
- Map<String, String> selectorConfig = context.getSubProperties(
- BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX);
-
- ChannelSelector selector = ChannelSelectorFactory.create(
- channels, selectorConfig);
-
- ChannelProcessor channelProcessor = new ChannelProcessor(selector);
- Configurables.configure(channelProcessor, context);
-
- source.setChannelProcessor(channelProcessor);
- conf.getSourceRunners().put(src,
- SourceRunner.forSource(source));
-
- }
- }
- }
-
- private void loadSinks(AgentConfiguration agentConf, NodeConfiguration conf)
- throws InstantiationException {
- Set<String> sinkNames = agentConf.getSinkSet();
- Map<String, ComponentConfiguration> compMap =
- agentConf.getSinkConfigMap();
- Map<String, Sink> sinks = new HashMap<String, Sink>();
- for (String sinkName : sinkNames) {
- ComponentConfiguration comp = compMap.get(sinkName);
- if(comp != null) {
- SinkConfiguration config = (SinkConfiguration) comp;
- Sink sink = getSinkFactory().create(comp.getComponentName(),
- comp.getType());
-
- Configurables.configure(sink, config);
-
- sink.setChannel(conf.getChannels().get(config.getChannel()));
- sinks.put(comp.getComponentName(), sink);
- }
- }
-
- Map<String, Context> sinkContexts = agentConf.getSinkContext();
- for (String sinkName : sinkNames) {
- Context context = sinkContexts.get(sinkName);
- if(context != null) {
- Sink sink = getSinkFactory().create(sinkName, context.getString(
- BasicConfigurationConstants.CONFIG_TYPE));
- Configurables.configure(sink, context);
-
- sink.setChannel(conf.getChannels().get(context.getString(
- BasicConfigurationConstants.CONFIG_CHANNEL)));
- sinks.put(sinkName, sink);
- }
- }
-
- loadSinkGroups(agentConf, sinks, conf);
- }
-
- private void loadSinkGroups(AgentConfiguration agentConf,
- Map<String, Sink> sinks, NodeConfiguration conf)
- throws InstantiationException {
- Set<String> sinkgroupNames = agentConf.getSinkgroupSet();
- Map<String, ComponentConfiguration> compMap =
- agentConf.getSinkGroupConfigMap();
- Map<String, String> usedSinks = new HashMap<String, String>();
- for (String groupName: sinkgroupNames) {
- ComponentConfiguration comp = compMap.get(groupName);
- if(comp != null) {
- SinkGroupConfiguration groupConf = (SinkGroupConfiguration) comp;
- List<String> groupSinkList = groupConf.getSinks();
- List<Sink> groupSinks = new ArrayList<Sink>();
- for (String sink : groupSinkList) {
- Sink s = sinks.remove(sink);
- if (s == null) {
- String sinkUser = usedSinks.get(sink);
- if (sinkUser != null) {
- throw new InstantiationException(String.format(
- "Sink %s of group %s already " +
- "in use by group %s", sink, groupName, sinkUser));
- } else {
- throw new InstantiationException(String.format(
- "Sink %s of group %s does "
- + "not exist or is not properly configured", sink,
- groupName));
- }
- }
- groupSinks.add(s);
- usedSinks.put(sink, groupName);
- }
- SinkGroup group = new SinkGroup(groupSinks);
- Configurables.configure(group, groupConf);
- conf.getSinkRunners().put(comp.getComponentName(),
- new SinkRunner(group.getProcessor()));
- }
- }
- // add any unasigned sinks to solo collectors
- for(Entry<String, Sink> entry : sinks.entrySet()) {
- if (!usedSinks.containsValue(entry.getKey())) {
- SinkProcessor pr = new DefaultSinkProcessor();
- List<Sink> sinkMap = new ArrayList<Sink>();
- sinkMap.add(entry.getValue());
- pr.setSinks(sinkMap);
- Configurables.configure(pr, new Context());
- conf.getSinkRunners().put(entry.getKey(),
- new SinkRunner(pr));
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/Application.java b/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
index 405afa3..e73423e 100644
--- a/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
+++ b/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
@@ -21,6 +21,11 @@ package org.apache.flume.node;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
@@ -29,150 +34,274 @@ import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
-import org.apache.flume.ChannelFactory;
+import org.apache.flume.Channel;
import org.apache.flume.Constants;
-import org.apache.flume.SinkFactory;
-import org.apache.flume.SourceFactory;
-import org.apache.flume.channel.DefaultChannelFactory;
-import org.apache.flume.conf.file.AbstractFileConfigurationProvider;
-import org.apache.flume.conf.properties.PropertiesFileConfigurationProvider;
-import org.apache.flume.lifecycle.LifecycleController;
-import org.apache.flume.lifecycle.LifecycleException;
+import org.apache.flume.Context;
+import org.apache.flume.SinkRunner;
+import org.apache.flume.SourceRunner;
+import org.apache.flume.instrumentation.MonitorService;
+import org.apache.flume.instrumentation.MonitoringType;
+import org.apache.flume.lifecycle.LifecycleAware;
import org.apache.flume.lifecycle.LifecycleState;
-import org.apache.flume.node.nodemanager.DefaultLogicalNodeManager;
-import org.apache.flume.sink.DefaultSinkFactory;
-import org.apache.flume.source.DefaultSourceFactory;
+import org.apache.flume.lifecycle.LifecycleSupervisor;
+import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
-public class Application {
+public class Application {
private static final Logger logger = LoggerFactory
.getLogger(Application.class);
- private String[] args;
- private File configurationFile;
- private String nodeName;
+ public static final String CONF_MONITOR_CLASS = "flume.monitoring.type";
+ public static final String CONF_MONITOR_PREFIX = "flume.monitoring.";
- private SourceFactory sourceFactory;
- private SinkFactory sinkFactory;
- private ChannelFactory channelFactory;
+ private final List<LifecycleAware> components;
+ private final LifecycleSupervisor supervisor;
+ private MaterializedConfiguration materializedConfiguration;
+ private MonitorService monitorServer;
- public static void main(String[] args) {
- Application application = new Application();
-
- application.setArgs(args);
+ public Application() {
+ this(new ArrayList<LifecycleAware>(0));
+ }
+ public Application(List<LifecycleAware> components) {
+ this.components = components;
+ supervisor = new LifecycleSupervisor();
+ }
- try {
- if (application.parseOptions()) {
- application.run();
- }
- } catch (ParseException e) {
- logger.error(e.getMessage());
- } catch (Exception e) {
- logger.error("A fatal error occurred while running. Exception follows.",
- e);
+ public void start() {
+ for(LifecycleAware component : components) {
+ supervisor.supervise(component,
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
}
}
- public Application() {
- sourceFactory = new DefaultSourceFactory();
- sinkFactory = new DefaultSinkFactory();
- channelFactory = new DefaultChannelFactory();
+
+ @Subscribe
+ public synchronized void handleConfigurationEvent(MaterializedConfiguration conf) {
+ stopAllComponents();
+ startAllComponents(conf);
}
- public boolean parseOptions() throws ParseException {
- Options options = new Options();
+ public void stop() {
+ supervisor.stop();
+ }
- Option option = new Option("n", "name", true, "the name of this node");
- options.addOption(option);
- option = new Option("f", "conf-file", true, "specify a conf file");
- options.addOption(option);
+ private void stopAllComponents() {
+ if (this.materializedConfiguration != null) {
+ logger.info("Shutting down configuration: {}", this.materializedConfiguration);
+ for (Entry<String, SourceRunner> entry : this.materializedConfiguration
+ .getSourceRunners().entrySet()) {
+ try{
+ logger.info("Stopping Source " + entry.getKey());
+ supervisor.unsupervise(entry.getValue());
+ } catch (Exception e){
+ logger.error("Error while stopping {}", entry.getValue(), e);
+ }
+ }
+
+ for (Entry<String, SinkRunner> entry :
+ this.materializedConfiguration.getSinkRunners().entrySet()) {
+ try{
+ logger.info("Stopping Sink " + entry.getKey());
+ supervisor.unsupervise(entry.getValue());
+ } catch (Exception e){
+ logger.error("Error while stopping {}", entry.getValue(), e);
+ }
+ }
+
+ for (Entry<String, Channel> entry :
+ this.materializedConfiguration.getChannels().entrySet()) {
+ try{
+ logger.info("Stopping Channel " + entry.getKey());
+ supervisor.unsupervise(entry.getValue());
+ } catch (Exception e){
+ logger.error("Error while stopping {}", entry.getValue(), e);
+ }
+ }
+ }
+ if(monitorServer != null) {
+ monitorServer.stop();
+ }
+ }
- option = new Option("h", "help", false, "display help text");
- options.addOption(option);
+ private void startAllComponents(MaterializedConfiguration materializedConfiguration) {
+ logger.info("Starting new configuration:{}", materializedConfiguration);
- CommandLineParser parser = new GnuParser();
- CommandLine commandLine = parser.parse(options, args);
+ this.materializedConfiguration = materializedConfiguration;
- if (commandLine.hasOption('f')) {
- configurationFile = new File(commandLine.getOptionValue('f'));
+ for (Entry<String, Channel> entry :
+ materializedConfiguration.getChannels().entrySet()) {
+ try{
+ logger.info("Starting Channel " + entry.getKey());
+ supervisor.supervise(entry.getValue(),
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ } catch (Exception e){
+ logger.error("Error while starting {}", entry.getValue(), e);
+ }
+ }
- if (!configurationFile.exists()) {
- // If command line invocation, then need to fail fast
- if (System.getProperty(Constants.SYSPROP_CALLED_FROM_SERVICE) == null) {
- String path = configurationFile.getPath();
- try {
- path = configurationFile.getCanonicalPath();
- } catch (IOException ex) {
- logger.error("Failed to read canonical path for file: " + path, ex);
- }
- throw new ParseException(
- "The specified configuration file does not exist: " + path);
+ /*
+ * Wait for all channels to start.
+ */
+ for(Channel ch: materializedConfiguration.getChannels().values()){
+ while(ch.getLifecycleState() != LifecycleState.START
+ && !supervisor.isComponentInErrorState(ch)){
+ try {
+ logger.info("Waiting for channel: " + ch.getName() +
+ " to start. Sleeping for 500 ms");
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ logger.error("Interrupted while waiting for channel to start.", e);
+ Throwables.propagate(e);
}
}
}
- if (commandLine.hasOption('n')) {
- nodeName = commandLine.getOptionValue('n');
+ for (Entry<String, SinkRunner> entry : materializedConfiguration.getSinkRunners()
+ .entrySet()) {
+ try{
+ logger.info("Starting Sink " + entry.getKey());
+ supervisor.supervise(entry.getValue(),
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ } catch (Exception e) {
+ logger.error("Error while starting {}", entry.getValue(), e);
+ }
}
- if (commandLine.hasOption('h')) {
- new HelpFormatter().printHelp("flume-ng node", options, true);
+ for (Entry<String, SourceRunner> entry : materializedConfiguration
+ .getSourceRunners().entrySet()) {
+ try{
+ logger.info("Starting Source " + entry.getKey());
+ supervisor.supervise(entry.getValue(),
+ new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
+ } catch (Exception e) {
+ logger.error("Error while starting {}", entry.getValue(), e);
+ }
+ }
+
+ this.loadMonitoring();
+ }
+
- return false;
+ @SuppressWarnings("unchecked")
+ private void loadMonitoring() {
+ Properties systemProps = System.getProperties();
+ Set<String> keys = systemProps.stringPropertyNames();
+ try {
+ if (keys.contains(CONF_MONITOR_CLASS)) {
+ String monitorType = systemProps.getProperty(CONF_MONITOR_CLASS);
+ Class<? extends MonitorService> klass;
+ try {
+ //Is it a known type?
+ klass = MonitoringType.valueOf(
+ monitorType.toUpperCase()).getMonitorClass();
+ } catch (Exception e) {
+ //Not a known type, use FQCN
+ klass = (Class<? extends MonitorService>) Class.forName(monitorType);
+ }
+ this.monitorServer = klass.newInstance();
+ Context context = new Context();
+ for (String key : keys) {
+ if (key.startsWith(CONF_MONITOR_PREFIX)) {
+ context.put(key.substring(CONF_MONITOR_PREFIX.length()),
+ systemProps.getProperty(key));
+ }
+ }
+ monitorServer.configure(context);
+ monitorServer.start();
+ }
+ } catch (Exception e) {
+ logger.warn("Error starting monitoring. "
+ + "Monitoring might not be available.", e);
}
- return true;
}
- public void run() throws LifecycleException, InterruptedException,
- InstantiationException {
+ public static void main(String[] args) {
- final FlumeNode node = new FlumeNode();
- DefaultLogicalNodeManager nodeManager = new DefaultLogicalNodeManager();
- AbstractFileConfigurationProvider configurationProvider =
- new PropertiesFileConfigurationProvider();
+ try {
- configurationProvider.setChannelFactory(channelFactory);
- configurationProvider.setSourceFactory(sourceFactory);
- configurationProvider.setSinkFactory(sinkFactory);
+ Options options = new Options();
- configurationProvider.setNodeName(nodeName);
- configurationProvider.setConfigurationAware(nodeManager);
- configurationProvider.setFile(configurationFile);
+ Option option = new Option("n", "name", true, "the name of this agent");
+ option.setRequired(true);
+ options.addOption(option);
- Preconditions.checkState(configurationFile != null,
- "Configuration file not specified");
- Preconditions.checkState(nodeName != null, "Node name not specified");
+ option = new Option("f", "conf-file", true, "specify a conf file");
+ option.setRequired(true);
+ options.addOption(option);
- node.setName(nodeName);
- node.setNodeManager(nodeManager);
- node.setConfigurationProvider(configurationProvider);
+ option = new Option("no-reload", false, "do not reload conf file if changed");
+ options.addOption(option);
- Runtime.getRuntime().addShutdownHook(new Thread("node-shutdownHook") {
+ option = new Option("h", "help", false, "display help text");
+ options.addOption(option);
- @Override
- public void run() {
- node.stop();
- }
+ CommandLineParser parser = new GnuParser();
+ CommandLine commandLine = parser.parse(options, args);
- });
+ File configurationFile = new File(commandLine.getOptionValue('f'));
+ String agentName = commandLine.getOptionValue('n');
+ boolean reload = !commandLine.hasOption("no-reload");
- node.start();
- LifecycleController.waitForOneOf(node, LifecycleState.START_OR_ERROR);
- LifecycleController.waitForOneOf(node, LifecycleState.STOP_OR_ERROR);
- }
+ if (commandLine.hasOption('h')) {
+ new HelpFormatter().printHelp("flume-ng agent", options, true);
+ return;
+ }
+ /*
+ * The following is to ensure that by default the agent
+ * will fail on startup if the file does not exist.
+ */
+ if (!configurationFile.exists()) {
+ // If command line invocation, then need to fail fast
+ if (System.getProperty(Constants.SYSPROP_CALLED_FROM_SERVICE) == null) {
+ String path = configurationFile.getPath();
+ try {
+ path = configurationFile.getCanonicalPath();
+ } catch (IOException ex) {
+ logger.error("Failed to read canonical path for file: " + path, ex);
+ }
+ throw new ParseException(
+ "The specified configuration file does not exist: " + path);
+ }
+ }
+ List<LifecycleAware> components = Lists.newArrayList();
+ Application application;
+ if(reload) {
+ EventBus eventBus = new EventBus(agentName + "-event-bus");
+ PollingPropertiesFileConfigurationProvider configurationProvider =
+ new PollingPropertiesFileConfigurationProvider(agentName,
+ configurationFile, eventBus, 30);
+ components.add(configurationProvider);
+ application = new Application(components);
+ eventBus.register(application);
+ } else {
+ PropertiesFileConfigurationProvider configurationProvider =
+ new PropertiesFileConfigurationProvider(agentName,
+ configurationFile);
+ application = new Application();
+ application.handleConfigurationEvent(configurationProvider.getConfiguration());
+ }
+ application.start();
- public String[] getArgs() {
- return args;
- }
+ final Application appReference = application;
+ Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
+ @Override
+ public void run() {
+ appReference.stop();
+ }
+ });
- public void setArgs(String[] args) {
- this.args = args;
+ } catch (Exception e) {
+ logger.error("A fatal error occurred while running. Exception follows.",
+ e);
+ }
}
-
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/ConfigurationProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/ConfigurationProvider.java b/flume-ng-node/src/main/java/org/apache/flume/node/ConfigurationProvider.java
index ae732aa..6a27898 100644
--- a/flume-ng-node/src/main/java/org/apache/flume/node/ConfigurationProvider.java
+++ b/flume-ng-node/src/main/java/org/apache/flume/node/ConfigurationProvider.java
@@ -19,8 +19,11 @@
package org.apache.flume.node;
-import org.apache.flume.lifecycle.LifecycleAware;
-public interface ConfigurationProvider extends LifecycleAware {
+public interface ConfigurationProvider {
+
+
+ public MaterializedConfiguration getConfiguration();
+
}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/FlumeNode.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/FlumeNode.java b/flume-ng-node/src/main/java/org/apache/flume/node/FlumeNode.java
deleted file mode 100644
index cee022d..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/node/FlumeNode.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.flume.node;
-
-import org.apache.flume.lifecycle.LifecycleAware;
-import org.apache.flume.lifecycle.LifecycleState;
-import org.apache.flume.lifecycle.LifecycleSupervisor;
-import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-
-public class FlumeNode implements LifecycleAware {
-
- private static final Logger logger = LoggerFactory.getLogger(FlumeNode.class);
-
- private String name;
- private LifecycleState lifecycleState;
- private NodeManager nodeManager;
- private ConfigurationProvider configurationProvider;
- private LifecycleSupervisor supervisor;
-
- public FlumeNode() {
- supervisor = new LifecycleSupervisor();
- }
-
- @Override
- public void start() {
-
- Preconditions.checkState(name != null, "Node name can not be null");
- Preconditions.checkState(nodeManager != null,
- "Node manager can not be null");
-
- supervisor.start();
-
- logger.info("Flume node starting - {}", name);
-
- supervisor.supervise(nodeManager,
- new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
- supervisor.supervise(configurationProvider,
- new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
-
- lifecycleState = LifecycleState.START;
- }
-
- @Override
- public void stop() {
-
- logger.info("Flume node stopping - {}", name);
-
- supervisor.stop();
-
- lifecycleState = LifecycleState.STOP;
- }
-
- @Override
- public String toString() {
- return "{ name:" + name + " nodeManager:" + nodeManager
- + " configurationProvider:" + configurationProvider + " }";
- }
-
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public NodeManager getNodeManager() {
- return nodeManager;
- }
-
- public void setNodeManager(NodeManager nodeManager) {
- this.nodeManager = nodeManager;
- }
-
- public ConfigurationProvider getConfigurationProvider() {
- return configurationProvider;
- }
-
- public void setConfigurationProvider(
- ConfigurationProvider configurationProvider) {
- this.configurationProvider = configurationProvider;
- }
-
- @Override
- public LifecycleState getLifecycleState() {
- return lifecycleState;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/NodeConfiguration.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/NodeConfiguration.java b/flume-ng-node/src/main/java/org/apache/flume/node/NodeConfiguration.java
deleted file mode 100644
index a24c939..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/node/NodeConfiguration.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.flume.node;
-
-import java.util.Map;
-
-import org.apache.flume.Channel;
-import org.apache.flume.SinkRunner;
-import org.apache.flume.SourceRunner;
-
-public interface NodeConfiguration {
-
- public Map<String, SourceRunner> getSourceRunners();
-
- public Map<String, SinkRunner> getSinkRunners();
-
- public Map<String, Channel> getChannels();
-
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/NodeManager.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/NodeManager.java b/flume-ng-node/src/main/java/org/apache/flume/node/NodeManager.java
deleted file mode 100644
index 7457d87..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/node/NodeManager.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.flume.node;
-
-import java.util.Set;
-
-import org.apache.flume.lifecycle.LifecycleAware;
-
-public interface NodeManager extends LifecycleAware {
-
- public boolean add(LifecycleAware node);
-
- public boolean remove(LifecycleAware node);
-
- public Set<LifecycleAware> getNodes();
-
- public void setNodes(Set<LifecycleAware> nodes);
-
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/e47fd1e7/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/AbstractLogicalNodeManager.java
----------------------------------------------------------------------
diff --git a/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/AbstractLogicalNodeManager.java b/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/AbstractLogicalNodeManager.java
deleted file mode 100644
index 1fda07b..0000000
--- a/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/AbstractLogicalNodeManager.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.flume.node.nodemanager;
-
-import java.util.HashSet;
-import java.util.Set;
-
-import org.apache.flume.lifecycle.LifecycleAware;
-import org.apache.flume.node.NodeManager;
-
-import com.google.common.base.Preconditions;
-
-abstract public class AbstractLogicalNodeManager implements NodeManager {
-
- private Set<LifecycleAware> nodes;
-
- public AbstractLogicalNodeManager() {
- nodes = new HashSet<LifecycleAware>();
- }
-
- @Override
- public boolean add(LifecycleAware node) {
- Preconditions.checkNotNull(node);
-
- return nodes.add(node);
- }
-
- @Override
- public boolean remove(LifecycleAware node) {
- Preconditions.checkNotNull(node);
-
- return nodes.remove(node);
- }
-
- @Override
- public Set<LifecycleAware> getNodes() {
- return nodes;
- }
-
- @Override
- public void setNodes(Set<LifecycleAware> nodes) {
- Preconditions.checkNotNull(nodes);
-
- this.nodes = new HashSet<LifecycleAware>(nodes);
- }
-
- @Override
- public String toString() {
- return "{ nodes:" + nodes + " }";
- }
-
-}