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/05/30 08:56:45 UTC
svn commit: r1344118 -
/incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java
Author: arvind
Date: Wed May 30 06:56:45 2012
New Revision: 1344118
URL: http://svn.apache.org/viewvc?rev=1344118&view=rev
Log:
FLUME-1236. Race condition between channel startup and sinks.
(Hari Shreedharan via Arvind Prabhakar)
Modified:
incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/node/nodemanager/DefaultLogicalNodeManager.java
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=1344118&r1=1344117&r2=1344118&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 Wed May 30 06:56:45 2012
@@ -33,6 +33,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
public class DefaultLogicalNodeManager extends AbstractLogicalNodeManager
implements NodeConfigurationAware {
@@ -104,6 +105,22 @@ public class DefaultLogicalNodeManager e
}
}
+ /*
+ * Wait for all channels to start.
+ */
+ for(Channel ch: nodeConfiguration.getChannels().values()){
+ while(ch.getLifecycleState() != LifecycleState.START){
+ 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);
+ }
+ }
+ }
+
for (Entry<String, SinkRunner> entry : nodeConfiguration.getSinkRunners()
.entrySet()) {
try{