flume-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From esam...@apache.org
Subject svn commit: r1157304 - in /incubator/flume/branches/flume-728/flume-ng-core/src: main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java
Date Sat, 13 Aug 2011 00:29:58 GMT
Author: esammer
Date: Sat Aug 13 00:29:57 2011
New Revision: 1157304

URL: http://svn.apache.org/viewvc?rev=1157304&view=rev
Log:
- Added support for "unsupervising" services as well as changing the desired
  state (which in turn effects a state transition by the monitor thread).
- Switched from using a fixed scheduled delay, we now schedule the next monitor
  run at the end of the current (i.e. chained scheduling). The reason we need
  to do this is because ScheduledExecutorService provides no (polite) method of
  unscheduling something so we go old school.
- Added hinkey unit tests for unsupervise() and setDesiredState() in
  LifecycleSupervisor.

Modified:
    incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
    incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java

Modified: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java?rev=1157304&r1=1157303&r2=1157304&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
(original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/lifecycle/LifecycleSupervisor.java
Sat Aug 13 00:29:57 2011
@@ -11,6 +11,7 @@ import org.apache.flume.Context;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 public class LifecycleSupervisor implements LifecycleAware {
@@ -101,6 +102,14 @@ public class LifecycleSupervisor impleme
   public synchronized void supervise(LifecycleAware lifecycleAware,
       SupervisorPolicy policy, LifecycleState desiredState) {
 
+    Preconditions.checkState(!supervisedProcesses.containsKey(lifecycleAware),
+        "Refusing to supervise " + lifecycleAware + " more than once");
+
+    if (logger.isDebugEnabled()) {
+      logger.debug("Supervising service:{} policy:{} desiredState:{}",
+          new Object[] { lifecycleAware, policy, desiredState });
+    }
+
     Supervisoree process = new Supervisoree();
     process.status = new Status();
 
@@ -110,9 +119,35 @@ public class LifecycleSupervisor impleme
     MonitorRunnable monitorRunnable = new MonitorRunnable();
     monitorRunnable.lifecycleAware = lifecycleAware;
     monitorRunnable.supervisoree = process;
+    monitorRunnable.monitorService = monitorService;
 
     supervisedProcesses.put(lifecycleAware, process);
-    monitorService.scheduleAtFixedRate(monitorRunnable, 0, 3, TimeUnit.SECONDS);
+    monitorService.schedule(monitorRunnable, 0, TimeUnit.SECONDS);
+  }
+
+  public synchronized void unsupervise(LifecycleAware lifecycleAware) {
+
+    Preconditions.checkState(supervisedProcesses.containsKey(lifecycleAware),
+        "Unaware of " + lifecycleAware + " - can not unsupervise");
+
+    logger.debug("Unsupervising service:{}", lifecycleAware);
+
+    Supervisoree supervisoree = supervisedProcesses.get(lifecycleAware);
+    supervisoree.status.discard = true;
+  }
+
+  public synchronized void setDesiredState(LifecycleAware lifecycleAware,
+      LifecycleState desiredState) {
+
+    Preconditions.checkState(supervisedProcesses.containsKey(lifecycleAware),
+        "Unaware of " + lifecycleAware + " - can not set desired state to "
+            + desiredState);
+
+    logger.debug("Setting desiredState:{} on service:{}", desiredState,
+        lifecycleAware);
+
+    Supervisoree supervisoree = supervisedProcesses.get(lifecycleAware);
+    supervisoree.status.desiredState = desiredState;
   }
 
   @Override
@@ -122,6 +157,7 @@ public class LifecycleSupervisor impleme
 
   public static class MonitorRunnable implements Runnable {
 
+    public ScheduledExecutorService monitorService;
     public LifecycleAware lifecycleAware;
     public Supervisoree supervisoree;
 
@@ -130,6 +166,11 @@ public class LifecycleSupervisor impleme
       logger.debug("checking process:{} supervisoree:{}", lifecycleAware,
           supervisoree);
 
+      if (supervisoree.status.discard) {
+        logger.debug("Halting monitoring on {}", supervisoree);
+        return;
+      }
+
       long now = System.currentTimeMillis();
 
       if (supervisoree.status.firstSeen == null) {
@@ -184,6 +225,8 @@ public class LifecycleSupervisor impleme
         }
       }
 
+      monitorService.schedule(this, 3, TimeUnit.SECONDS);
+
       logger.debug("Status check complete");
     }
 
@@ -195,12 +238,13 @@ public class LifecycleSupervisor impleme
     public LifecycleState lastSeenState;
     public LifecycleState desiredState;
     public int failures;
+    public boolean discard;
 
     @Override
     public String toString() {
       return "{ lastSeen:" + lastSeen + " lastSeenState:" + lastSeenState
           + " desiredState:" + desiredState + " firstSeen:" + firstSeen
-          + " failures:" + failures + " }";
+          + " failures:" + failures + " discard:" + discard + " }";
     }
 
   }

Modified: incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java?rev=1157304&r1=1157303&r2=1157304&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java
(original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/lifecycle/TestLifecycleSupervisor.java
Sat Aug 13 00:29:57 2011
@@ -1,10 +1,12 @@
 package org.apache.flume.lifecycle;
 
 import org.apache.flume.Context;
+import org.apache.flume.CounterGroup;
 import org.apache.flume.LogicalNode;
 import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy;
 import org.apache.flume.sink.NullSink;
 import org.apache.flume.source.SequenceGeneratorSource;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -105,4 +107,86 @@ public class TestLifecycleSupervisor {
     supervisor.stop(context);
   }
 
+  @Test
+  public void testUnsuperviseServce() throws LifecycleException,
+      InterruptedException {
+    Context context = new Context();
+
+    supervisor.start(context);
+
+    LifecycleAware service = new CountingLifecycleAware();
+    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
+
+    supervisor.supervise(service, policy, LifecycleState.START);
+    supervisor.unsupervise(service);
+
+    service.stop(context);
+
+    supervisor.stop(context);
+  }
+
+  @Test
+  public void testStopServce() throws LifecycleException, InterruptedException {
+    Context context = new Context();
+
+    supervisor.start(context);
+
+    CountingLifecycleAware service = new CountingLifecycleAware();
+    SupervisorPolicy policy = new SupervisorPolicy.OnceOnlyPolicy();
+
+    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("start"));
+    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));
+
+    supervisor.supervise(service, policy, LifecycleState.START);
+
+    Thread.sleep(3200);
+
+    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
+    Assert.assertEquals(Long.valueOf(0), service.counterGroup.get("stop"));
+
+    supervisor.setDesiredState(service, LifecycleState.STOP);
+
+    Thread.sleep(3200);
+
+    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("start"));
+    Assert.assertEquals(Long.valueOf(1), service.counterGroup.get("stop"));
+
+    supervisor.stop(context);
+  }
+
+  public static class CountingLifecycleAware implements LifecycleAware {
+
+    public CounterGroup counterGroup;
+    private LifecycleState lifecycleState;
+
+    public CountingLifecycleAware() {
+      lifecycleState = LifecycleState.IDLE;
+      counterGroup = new CounterGroup();
+    }
+
+    @Override
+    public void start(Context context) throws LifecycleException,
+        InterruptedException {
+
+      counterGroup.incrementAndGet("start");
+
+      lifecycleState = LifecycleState.START;
+    }
+
+    @Override
+    public void stop(Context context) throws LifecycleException,
+        InterruptedException {
+
+      counterGroup.incrementAndGet("stop");
+
+      lifecycleState = LifecycleState.STOP;
+    }
+
+    @Override
+    public LifecycleState getLifecycleState() {
+      return lifecycleState;
+    }
+
+  }
+
 }



Mime
View raw message