Repository: apache/fluo
Issue: MiniAccismus.waitForObservers() returns before processing is finished

In MiniAccismus, the waitForObservers() method is occasionally returning before all observers are done processing.  This is because the method returns if there a no notifications available in the table.  However, observers could still be running and creating new notifications.  

The method should instead check that no observers are running and no new notifications are in the table before returning.


Changed files:
modules/core/src/main/java/accismus/impl/WorkerTask.java, modules/core/src/main/java/accismus/api/test/MiniAccismus.java

diff --git a/modules/core/src/main/java/accismus/api/test/MiniAccismus.java b/modules/core/src/main/java/accismus/api/test/MiniAccismus.java
index e89fa064..0a25d75e 100644
--- a/modules/core/src/main/java/accismus/api/test/MiniAccismus.java
+++ b/modules/core/src/main/java/accismus/api/test/MiniAccismus.java
@@ -40,6 +40,33 @@ public class MiniAccismus {
   private ExecutorService tp;
   private AtomicBoolean shutdownFlag;
 
+  private int numProcessing = 0;
+
+  private class MiniWorkerTask extends WorkerTask {
+
+    public MiniWorkerTask(Configuration config, AtomicBoolean shutdownFlag) {
+      super(config, shutdownFlag);
+    }
+
+    @Override
+    public void startedProcessing() {
+      synchronized (MiniAccismus.this) {
+        numProcessing++;
+      }
+    }
+
+    @Override
+    public void finishedProcessing(long numProcessed) {
+      synchronized (MiniAccismus.this) {
+        numProcessing--;
+      }
+    }
+  }
+
+  private synchronized boolean isProcessing(Scanner scanner) {
+    return scanner.iterator().hasNext() || numProcessing > 0;
+  }
+
   public MiniAccismus(Properties props) {
     try {
       aconfig = new Configuration(props);
@@ -59,7 +86,7 @@ public class MiniAccismus {
 
       tp = Executors.newFixedThreadPool(numThreads);
       for (int i = 0; i < numThreads; i++) {
-        tp.submit(new WorkerTask(aconfig, shutdownFlag));
+        tp.submit(new MiniWorkerTask(aconfig, shutdownFlag));
       }
 
     } catch (Exception e) {
@@ -85,14 +112,14 @@ public class MiniAccismus {
   }
 
   public void waitForObservers() {
-    // TODO create a better implementation
     try {
       Scanner scanner = aconfig.getConnector().createScanner(aconfig.getTable(), aconfig.getAuthorizations());
       scanner.fetchColumnFamily(ByteUtil.toText(Constants.NOTIFY_CF));
 
-      while (scanner.iterator().hasNext()) {
+      while (isProcessing(scanner)) {
         Thread.sleep(100);
       }
+
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
diff --git a/modules/core/src/main/java/accismus/impl/WorkerTask.java b/modules/core/src/main/java/accismus/impl/WorkerTask.java
index 190fe49a..fe2abc39 100644
--- a/modules/core/src/main/java/accismus/impl/WorkerTask.java
+++ b/modules/core/src/main/java/accismus/impl/WorkerTask.java
@@ -63,9 +63,12 @@ public class WorkerTask implements Runnable {
       while (!shutdownFlag.get()) {
         long numProcessed = 0;
         try {
+          startedProcessing();
           numProcessed = worker.processUpdates(colObservers);
         } catch (Exception e) {
           log.error("Error while processing updates", e);
+        } finally {
+          finishedProcessing(numProcessed);
         }
 
         if (numProcessed > 0)
@@ -92,4 +95,8 @@ public class WorkerTask implements Runnable {
     }
   }
 
+  public void startedProcessing() {}
+
+  public void finishedProcessing(long numProcessed) {}
+
 }