You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by ar...@apache.org on 2012/03/23 21:58:24 UTC
svn commit: r1304595 - in /incubator/flume/trunk:
flume-ng-core/src/main/java/org/apache/flume/lifecycle/
flume-ng-core/src/main/java/org/apache/flume/source/
flume-ng-node/src/main/java/org/apache/flume/conf/properties/
flume-ng-node/src/main/java/org...
Author: arvind
Date: Fri Mar 23 20:58:23 2012
New Revision: 1304595
URL: http://svn.apache.org/viewvc?rev=1304595&view=rev
Log:
FLUME-1036. Reconfiguration of AVRO or NETCAT source causes port bind exception
(Hari Shreedharan via Arvind Prabhakar)
Modified:
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java
incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java?rev=1304595&r1=1304594&r2=1304595&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java Fri Mar 23 20:58:23 2012
@@ -147,8 +147,14 @@ public class LifecycleSupervisor impleme
logger.debug("Unsupervising service:{}", lifecycleAware);
+ synchronized (lifecycleAware) {
Supervisoree supervisoree = supervisedProcesses.get(lifecycleAware);
supervisoree.status.discard = true;
+ this.setDesiredState(lifecycleAware, LifecycleState.STOP);
+ logger.info("Stopping component: {}", lifecycleAware);
+ lifecycleAware.stop();
+ }
+ supervisedProcesses.remove(lifecycleAware);
}
public synchronized void setDesiredState(LifecycleAware lifecycleAware,
@@ -190,6 +196,13 @@ public class LifecycleSupervisor impleme
}
supervisoree.status.lastSeen = now;
+ synchronized (lifecycleAware) {
+ if (supervisoree.status.discard) {
+ // Unsupervise has already been called on this.
+ logger.info("Component has already been stopped {}", lifecycleAware);
+ return;
+ }
+
supervisoree.status.lastSeenState = lifecycleAware.getLifecycleState();
if (!lifecycleAware.getLifecycleState().equals(
@@ -226,12 +239,14 @@ public class LifecycleSupervisor impleme
supervisoree.status.desiredState);
}
- if (!supervisoree.policy.isValid(lifecycleAware, supervisoree.status)) {
+ if (!supervisoree.policy.isValid(
+ lifecycleAware, supervisoree.status)) {
logger.error(
"Policy {} of {} has been violated - supervisor should exit!",
supervisoree.policy, lifecycleAware);
}
}
+ }
if (!supervisoree.status.discard) {
monitorService.schedule(this, 3, TimeUnit.SECONDS);
@@ -241,7 +256,6 @@ public class LifecycleSupervisor impleme
logger.debug("Status check complete");
}
-
}
public static class Status {
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java?rev=1304595&r1=1304594&r2=1304595&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/NetcatSource.java Fri Mar 23 20:58:23 2012
@@ -114,7 +114,7 @@ public class NetcatSource extends Abstra
port = 0;
counterGroup = new CounterGroup();
- acceptThreadShouldStop = new AtomicBoolean();
+ acceptThreadShouldStop = new AtomicBoolean(false);
}
@Override
@@ -152,7 +152,7 @@ public class NetcatSource extends Abstra
}
AcceptHandler acceptRunnable = new AcceptHandler();
-
+ acceptThreadShouldStop.set(false);
acceptRunnable.counterGroup = counterGroup;
acceptRunnable.handlerService = handlerService;
acceptRunnable.shouldStop = acceptThreadShouldStop;
Modified: incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java?rev=1304595&r1=1304594&r2=1304595&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java (original)
+++ incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java Fri Mar 23 20:58:23 2012
@@ -189,7 +189,7 @@ import org.slf4j.LoggerFactory;
* @see java.util.Properties#load(java.io.Reader)
*/
public class PropertiesFileConfigurationProvider extends
-AbstractFileConfigurationProvider {
+ AbstractFileConfigurationProvider {
private static final Logger LOGGER = LoggerFactory
.getLogger(PropertiesFileConfigurationProvider.class);
@@ -203,6 +203,7 @@ AbstractFileConfigurationProvider {
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);
@@ -254,7 +255,8 @@ AbstractFileConfigurationProvider {
}
}
- private void loadSources(AgentConfiguration agentConf, NodeConfiguration conf)
+ private void
+ loadSources(AgentConfiguration agentConf, NodeConfiguration conf)
throws InstantiationException {
for (ComponentConfiguration comp : agentConf.getSources()) {
@@ -278,7 +280,7 @@ AbstractFileConfigurationProvider {
channels.add(conf.getChannels().get(chName));
}
- Map<String,String> selectorConfig = comp.getSubconfiguration("selector");
+ Map<String, String> selectorConfig = comp.getSubconfiguration("selector");
ChannelSelector selector = ChannelSelectorFactory.create(
channels, selectorConfig);
@@ -299,7 +301,6 @@ AbstractFileConfigurationProvider {
Context context = new Context();
Map<String, String> componentConfig = comp.getConfiguration();
-
Sink sink = getSinkFactory().create(comp.getComponentName(),
componentConfig.get("type"));
@@ -331,12 +332,12 @@ AbstractFileConfigurationProvider {
String groupSinkList = groupConf.get("sinks");
StringTokenizer sinkTokenizer = new StringTokenizer(groupSinkList, " \t");
List<Sink> groupSinks = new ArrayList<Sink>();
- while(sinkTokenizer.hasMoreTokens()) {
+ while (sinkTokenizer.hasMoreTokens()) {
String sinkName = sinkTokenizer.nextToken();
Sink s = sinks.remove(sinkName);
- if(s == null) {
+ if (s == null) {
String sinkUser = usedSinks.get(sinkName);
- if(sinkUser != null) {
+ if (sinkUser != null) {
throw new InstantiationException(String.format(
"Sink %s of group %s already " +
"in use by group %s", sinkName, groupName, sinkUser));
@@ -356,7 +357,7 @@ AbstractFileConfigurationProvider {
new SinkRunner(group.getProcessor()));
}
// add any unasigned sinks to solo collectors
- for(Entry<String, Sink> entry : sinks.entrySet()) {
+ for (Entry<String, Sink> entry : sinks.entrySet()) {
if (!usedSinks.containsValue(entry.getKey())) {
SinkProcessor pr = new DefaultSinkProcessor();
List<Sink> sinkMap = new ArrayList<Sink>();
Modified: incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java?rev=1304595&r1=1304594&r2=1304595&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java (original)
+++ incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java Fri Mar 23 20:58:23 2012
@@ -41,16 +41,33 @@ public class DefaultLogicalNodeManager e
private LifecycleSupervisor nodeSupervisor;
private LifecycleState lifecycleState;
+ private NodeConfiguration nodeConfiguration;
public DefaultLogicalNodeManager() {
nodeSupervisor = new LifecycleSupervisor();
lifecycleState = LifecycleState.IDLE;
+ nodeConfiguration = null;
}
@Override
public void onNodeConfigurationChanged(NodeConfiguration nodeConfiguration) {
logger.info("Node configuration change:{}", nodeConfiguration);
+ if (this.nodeConfiguration != null) {
+ logger
+ .info("Shutting down old configuration: {}", this.nodeConfiguration);
+ for (Entry<String, SinkRunner> entry : nodeConfiguration.getSinkRunners()
+ .entrySet()) {
+ nodeSupervisor.unsupervise(entry.getValue());
+ }
+
+ for (Entry<String, SourceRunner> entry : nodeConfiguration
+ .getSourceRunners().entrySet()) {
+ nodeSupervisor.unsupervise(entry.getValue());
+ }
+ }
+
+ this.nodeConfiguration = nodeConfiguration;
for (Entry<String, SinkRunner> entry : nodeConfiguration.getSinkRunners()
.entrySet()) {
@@ -95,7 +112,6 @@ public class DefaultLogicalNodeManager e
"You can not remove nodes from a manager that hasn't been started");
if (super.remove(node)) {
- nodeSupervisor.setDesiredState(node, LifecycleState.STOP);
nodeSupervisor.unsupervise(node);
return true;