fluo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject incubator-fluo git commit: fixes #500 Made scanning for notifications scalable
Date Fri, 21 Apr 2017 02:13:26 GMT
Repository: incubator-fluo
Updated Branches:
  refs/heads/master 2e8932670 -> 3da76dcf0


fixes #500 Made scanning for notifications scalable


Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo/commit/3da76dcf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo/tree/3da76dcf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo/diff/3da76dcf

Branch: refs/heads/master
Commit: 3da76dcf06702dd9cd2e9b04fd9990cc0be5b151
Parents: 2e89326
Author: Keith Turner <kturner@apache.org>
Authored: Fri Apr 7 18:48:27 2017 -0400
Committer: Keith Turner <kturner@apache.org>
Committed: Thu Apr 20 22:12:01 2017 -0400

----------------------------------------------------------------------
 .../fluo/core/impl/FluoConfigurationImpl.java   |  16 +-
 .../core/worker/NotificationFinderFactory.java  |   4 +-
 .../fluo/core/worker/NotificationProcessor.java |  89 ++++-
 .../fluo/core/worker/TabletInfoCache.java       | 157 --------
 .../finder/hash/HashNotificationFinder.java     | 204 ----------
 .../core/worker/finder/hash/ModulusParams.java  |  28 --
 .../worker/finder/hash/ParitionManager.java     | 372 +++++++++++++++++++
 .../core/worker/finder/hash/PartitionInfo.java  |  98 +++++
 .../hash/PartitionInfoChangedException.java     |  20 +
 .../hash/PartitionNotificationFinder.java       |  84 +++++
 .../fluo/core/worker/finder/hash/RangeSet.java  |  88 +++++
 .../fluo/core/worker/finder/hash/ScanTask.java  | 133 ++++---
 .../worker/finder/hash/SerializedSplits.java    | 119 ++++++
 .../core/worker/finder/hash/TableRange.java     | 113 ++++++
 .../fluo/core/worker/finder/hash/HashTest.java  |   2 +-
 .../finder/hash/PartitionManagerTest.java       | 103 +++++
 .../finder/hash/SerializedSplitsTest.java       |  79 ++++
 .../core/worker/finder/hash/TableRangeTest.java | 125 +++++++
 .../apache/fluo/integration/impl/WorkerIT.java  |   6 +-
 .../java/org/apache/fluo/mini/MiniFluoImpl.java |   4 +-
 20 files changed, 1358 insertions(+), 486 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/impl/FluoConfigurationImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/FluoConfigurationImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/FluoConfigurationImpl.java
index a9f5fd4..9af6feb 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/FluoConfigurationImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/FluoConfigurationImpl.java
@@ -16,7 +16,6 @@
 package org.apache.fluo.core.impl;
 
 import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.core.worker.finder.hash.ScanTask;
 
 /**
  * Contains implementation-related Fluo properties that should not be exposed in the API in
@@ -28,13 +27,16 @@ public class FluoConfigurationImpl {
 
   public static final String ORACLE_PORT_PROP = FLUO_IMPL_PREFIX + ".oracle.port";
   public static final String WORKER_FINDER_PROP = FLUO_IMPL_PREFIX + ".worker.finder";
+  public static final String WORKER_PARTITION_GROUP_SIZE = FLUO_IMPL_PREFIX
+      + ".worker.finder.partition.groupSize";
+  public static final int WORKER_PARTITION_GROUP_SIZE_DEFAULT = 7;
   public static final String METRICS_RESERVOIR_PROP = FLUO_IMPL_PREFIX + ".metrics.reservoir";
-  public static final String MIN_SLEEP_TIME_PROP = FLUO_IMPL_PREFIX
-      + ScanTask.class.getSimpleName() + ".minSleep";
-  public static final int MIN_SLEEP_TIME_DEFAULT = 5000;
-  public static final String MAX_SLEEP_TIME_PROP = FLUO_IMPL_PREFIX
-      + ScanTask.class.getSimpleName() + ".maxSleep";
-  public static final int MAX_SLEEP_TIME_DEFAULT = 5 * 60 * 1000;
+  public static final String NTFY_FINDER_MIN_SLEEP_TIME_PROP = FLUO_IMPL_PREFIX
+      + ".worker.finder.minSleep";
+  public static final int NTFY_FINDER_MIN_SLEEP_TIME_DEFAULT = 5000;
+  public static final String NTFY_FINDER_MAX_SLEEP_TIME_PROP = FLUO_IMPL_PREFIX
+      + ".worker.finder.maxSleep";
+  public static final int NTFY_FINDER_MAX_SLEEP_TIME_DEFAULT = 5 * 60 * 1000;
 
   // Time period that each client will update ZK with their oldest active timestamp
   // If period is too short, Zookeeper may be overloaded. If too long, garbage collection

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationFinderFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationFinderFactory.java b/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationFinderFactory.java
index 92b0382..3938c07 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationFinderFactory.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationFinderFactory.java
@@ -17,13 +17,13 @@ package org.apache.fluo.core.worker;
 
 import org.apache.fluo.api.config.FluoConfiguration;
 import org.apache.fluo.core.impl.FluoConfigurationImpl;
-import org.apache.fluo.core.worker.finder.hash.HashNotificationFinder;
+import org.apache.fluo.core.worker.finder.hash.PartitionNotificationFinder;
 
 public class NotificationFinderFactory {
   public static NotificationFinder newNotificationFinder(FluoConfiguration conf) {
     String clazz =
         conf.getString(FluoConfigurationImpl.WORKER_FINDER_PROP,
-            HashNotificationFinder.class.getName());
+            PartitionNotificationFinder.class.getName());
     try {
       return Class.forName(clazz).asSubclass(NotificationFinder.class).newInstance();
     } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationProcessor.java b/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationProcessor.java
index 56aa78d..3b6bba5 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationProcessor.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/NotificationProcessor.java
@@ -16,13 +16,17 @@
 package org.apache.fluo.core.worker;
 
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 
 import com.codahale.metrics.Gauge;
 import org.apache.fluo.api.data.Column;
@@ -64,8 +68,12 @@ public class NotificationProcessor implements AutoCloseable {
   // little utility class that tracks all notifications in queue
   private class NotificationTracker {
     private Map<RowColumn, Future<?>> queuedWork = new HashMap<>();
+    private Set<RowColumn> recentlyDeleted = new HashSet<>();
     private long sizeInBytes = 0;
+    private Map<Long, Predicate<RowColumn>> memoryPredicates = new HashMap<>();
+    private Predicate<RowColumn> memoryPredicate = rc -> false;
     private static final long MAX_SIZE = 1 << 24;
+    private long nextSessionId = 0;
 
     private long size(RowColumn rowCol) {
       Column col = rowCol.getColumn();
@@ -75,7 +83,7 @@ public class NotificationProcessor implements AutoCloseable {
 
     public synchronized boolean add(RowColumn rowCol, Future<?> task) {
 
-      if (queuedWork.containsKey(rowCol)) {
+      if (queuedWork.containsKey(rowCol) || recentlyDeleted.contains(rowCol)) {
         return false;
       }
 
@@ -98,6 +106,9 @@ public class NotificationProcessor implements AutoCloseable {
 
     public synchronized void remove(RowColumn rowCol) {
       if (queuedWork.remove(rowCol) != null) {
+        if (memoryPredicate.test(rowCol)) {
+          recentlyDeleted.add(rowCol);
+        }
         sizeInBytes -= size(rowCol);
         notify();
       }
@@ -123,6 +134,34 @@ public class NotificationProcessor implements AutoCloseable {
       return true;
     }
 
+    private void resetMemoryPredicate() {
+      memoryPredicate = null;
+      for (Predicate<RowColumn> p : this.memoryPredicates.values()) {
+        if (memoryPredicate == null) {
+          memoryPredicate = p;
+        } else {
+          memoryPredicate = p.or(memoryPredicate);
+        }
+      }
+    }
+
+    public synchronized long beginAddingNotifications(Predicate<RowColumn> memoryPredicate) {
+      long sessionId = nextSessionId++;
+      this.memoryPredicates.put(sessionId, Objects.requireNonNull(memoryPredicate));
+      resetMemoryPredicate();
+      return sessionId;
+    }
+
+    public synchronized void finishAddingNotifications(long sessionId) {
+      this.memoryPredicates.remove(sessionId);
+      if (memoryPredicates.size() == 0) {
+        recentlyDeleted.clear();
+        memoryPredicate = rc -> false;
+      } else {
+        resetMemoryPredicate();
+      }
+    }
+
   }
 
   private static class NotificationProcessingTask implements Runnable {
@@ -176,25 +215,47 @@ public class NotificationProcessor implements AutoCloseable {
     }
   }
 
-  public boolean addNotification(final NotificationFinder notificationFinder,
-      final Notification notification) {
+  public class Session implements AutoCloseable {
+    private long id;
 
-    WorkTaskAsync workTask =
-        new WorkTaskAsync(this, notificationFinder, env, notification, observers);
-    FutureTask<?> ft = new FutureNotificationTask(notification, notificationFinder, workTask);
+    public Session(Predicate<RowColumn> memoryPredicate) {
+      this.id = tracker.beginAddingNotifications(memoryPredicate);
+    }
 
-    if (!tracker.add(notification.getRowColumn(), ft)) {
-      return false;
+    public boolean addNotification(final NotificationFinder notificationFinder,
+        final Notification notification) {
+
+      WorkTaskAsync workTask =
+          new WorkTaskAsync(NotificationProcessor.this, notificationFinder, env, notification,
+              observers);
+      FutureTask<?> ft = new FutureNotificationTask(notification, notificationFinder, workTask);
+
+      if (!tracker.add(notification.getRowColumn(), ft)) {
+        return false;
+      }
+
+      try {
+        executor.execute(ft);
+      } catch (RejectedExecutionException rje) {
+        tracker.remove(notification.getRowColumn());
+        throw rje;
+      }
+
+      return true;
     }
 
-    try {
-      executor.execute(ft);
-    } catch (RejectedExecutionException rje) {
-      tracker.remove(notification.getRowColumn());
-      throw rje;
+    public void close() {
+      tracker.finishAddingNotifications(id);
     }
+  }
 
-    return true;
+  /**
+   * Starts a session for adding notifications. During this session, any notifications that are
+   * deleted and match the predicate will be remembered. These remembered notifications can not be
+   * added again while the session is active.
+   */
+  public Session beginAddingNotifications(Predicate<RowColumn> memoryPredicate) {
+    return new Session(memoryPredicate);
   }
 
   public void requeueNotification(final NotificationFinder notificationFinder,

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/TabletInfoCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/TabletInfoCache.java b/modules/core/src/main/java/org/apache/fluo/core/worker/TabletInfoCache.java
deleted file mode 100644
index 4d17965..0000000
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/TabletInfoCache.java
+++ /dev/null
@@ -1,157 +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.fluo.core.worker;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Supplier;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.data.Range;
-import org.apache.fluo.core.impl.Environment;
-import org.apache.hadoop.io.Text;
-
-public class TabletInfoCache<T, S extends Supplier<T>> {
-  private static final long CACHE_TIME = 5 * 60 * 1000;
-
-  private final Environment env;
-  private List<TabletInfo<T>> cachedTablets;
-  private long listSplitsTime = 0;
-  private S supplier;
-
-  public static class TabletInfo<T> {
-    private final Text start;
-    private final Text end;
-    private T data;
-
-    TabletInfo(Text start, Text end, T data) {
-      this.start = start;
-      this.end = end;
-      this.data = data;
-    }
-
-    private int hashCode(Text t) {
-      if (t == null) {
-        return 0;
-      }
-      return t.hashCode();
-    }
-
-    @Override
-    public int hashCode() {
-      return hashCode(start) + hashCode(end);
-    }
-
-    private boolean equals(Text t1, Text t2) {
-      if (t1 == t2) {
-        return true;
-      }
-
-      if (t1 == null || t2 == null) {
-        return false;
-      }
-
-      return t1.equals(t2);
-    }
-
-    @Override
-    public boolean equals(Object o) {
-      if (o instanceof TabletInfo) {
-        @SuppressWarnings("rawtypes")
-        TabletInfo oti = (TabletInfo) o;
-
-        if (equals(start, oti.start)) {
-          return equals(end, oti.end);
-        }
-
-        return false;
-      }
-
-      return false;
-    }
-
-    public Text getStart() {
-      return start;
-    }
-
-    public Text getEnd() {
-      return end;
-    }
-
-    public T getData() {
-      return data;
-    }
-
-    public Range getRange() {
-      return new Range(start, false, end, true);
-    }
-  }
-
-  public TabletInfoCache(Environment env, S supplier) {
-    this.env = env;
-    this.supplier = supplier;
-  }
-
-  private List<TabletInfo<T>> listSplits() throws TableNotFoundException,
-      AccumuloSecurityException, AccumuloException {
-    List<Text> splits =
-        new ArrayList<>(env.getConnector().tableOperations().listSplits(env.getTable()));
-    Collections.sort(splits);
-
-    List<TabletInfo<T>> tablets = new ArrayList<>(splits.size() + 1);
-    for (int i = 0; i < splits.size(); i++) {
-      tablets
-          .add(new TabletInfo<>(i == 0 ? null : splits.get(i - 1), splits.get(i), supplier.get()));
-    }
-
-    tablets.add(new TabletInfo<>(splits.size() == 0 ? null : splits.get(splits.size() - 1), null,
-        supplier.get()));
-    listSplitsTime = System.currentTimeMillis();
-    return tablets;
-  }
-
-  public synchronized List<TabletInfo<T>> getTablets() throws Exception {
-    if (cachedTablets == null) {
-      cachedTablets = listSplits();
-    } else if (System.currentTimeMillis() - listSplitsTime > CACHE_TIME) {
-      List<TabletInfo<T>> tablets = listSplits();
-      Map<TabletInfo<T>, TabletInfo<T>> oldTablets = new HashMap<>();
-      for (TabletInfo<T> tabletInfo : cachedTablets) {
-        oldTablets.put(tabletInfo, tabletInfo);
-      }
-
-      List<TabletInfo<T>> newTablets = new ArrayList<>(tablets.size());
-
-      for (TabletInfo<T> tabletInfo : tablets) {
-        TabletInfo<T> oldTI = oldTablets.get(tabletInfo);
-        if (oldTI != null) {
-          newTablets.add(oldTI);
-        } else {
-          newTablets.add(tabletInfo);
-        }
-      }
-
-      cachedTablets = newTablets;
-    }
-
-    return Collections.unmodifiableList(cachedTablets);
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/HashNotificationFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/HashNotificationFinder.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/HashNotificationFinder.java
deleted file mode 100644
index e18ec2f..0000000
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/HashNotificationFinder.java
+++ /dev/null
@@ -1,204 +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.fluo.core.worker.finder.hash;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import org.apache.accumulo.core.data.ArrayByteSequence;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
-import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
-import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode;
-import org.apache.curator.utils.ZKPaths;
-import org.apache.fluo.accumulo.iterators.NotificationHashFilter;
-import org.apache.fluo.accumulo.util.NotificationUtil;
-import org.apache.fluo.accumulo.util.ZookeeperPath;
-import org.apache.fluo.core.impl.Environment;
-import org.apache.fluo.core.impl.Notification;
-import org.apache.fluo.core.util.ByteUtil;
-import org.apache.fluo.core.util.UtilWaitThread;
-import org.apache.fluo.core.worker.NotificationFinder;
-import org.apache.fluo.core.worker.NotificationProcessor;
-import org.apache.fluo.core.worker.TxResult;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class HashNotificationFinder implements NotificationFinder {
-
-  private NotificationProcessor notificationProcessor;
-  private CuratorFramework curator;
-  private List<String> finders = Collections.emptyList();
-  private int updates = 0;
-  private ModulusParams modParams;
-  private Environment env;
-  private AtomicBoolean stopped = new AtomicBoolean(false);
-
-  private Thread scanThread;
-  private PathChildrenCache childrenCache;
-  private PersistentEphemeralNode myESNode;
-
-  private static final Logger log = LoggerFactory.getLogger(HashNotificationFinder.class);
-
-  static class ModParamsChangedException extends RuntimeException {
-    private static final long serialVersionUID = 1L;
-  }
-
-  private class FindersListener implements PathChildrenCacheListener {
-
-    @Override
-    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
-      switch (event.getType()) {
-        case CHILD_ADDED:
-        case CHILD_REMOVED:
-          if (!stopped.get()) {
-            updateFinders();
-          }
-          break;
-        case CHILD_UPDATED:
-          log.warn("unexpected event " + event);
-          break;
-        default:
-          break;
-
-      }
-    }
-
-  }
-
-  private synchronized void updateFinders() {
-
-    String me = myESNode.getActualPath();
-    while (me == null) {
-      UtilWaitThread.sleep(100);
-      me = myESNode.getActualPath();
-    }
-    me = ZKPaths.getNodeFromPath(me);
-
-    List<String> children = new ArrayList<>();
-    for (ChildData childData : childrenCache.getCurrentData()) {
-      children.add(ZKPaths.getNodeFromPath(childData.getPath()));
-    }
-
-    Collections.sort(children);
-
-    if (!finders.equals(children)) {
-      int index = children.indexOf(me);
-      if (index == -1) {
-        this.modParams = null;
-        finders = Collections.emptyList();
-        log.debug("Did not find self in list of finders " + me);
-      } else {
-        updates++;
-        this.modParams = new ModulusParams(children.indexOf(me), children.size(), updates);
-        finders = children;
-        log.debug("updated modulus params " + modParams.remainder + " " + modParams.divisor);
-      }
-    }
-  }
-
-  synchronized ModulusParams getModulusParams() {
-    return modParams;
-  }
-
-  @Override
-  public void init(Environment env, NotificationProcessor notificationProcessor) {
-    Preconditions.checkState(this.notificationProcessor == null);
-
-    this.notificationProcessor = notificationProcessor;
-
-    this.env = env;
-    this.curator = env.getSharedResources().getCurator();
-
-    try {
-      myESNode =
-          new PersistentEphemeralNode(curator, Mode.EPHEMERAL_SEQUENTIAL, ZookeeperPath.FINDERS
-              + "/f-", new byte[0]);
-      myESNode.start();
-      myESNode.waitForInitialCreate(1, TimeUnit.MINUTES);
-
-      childrenCache =
-          new PathChildrenCache(env.getSharedResources().getCurator(), ZookeeperPath.FINDERS, false);
-      childrenCache.getListenable().addListener(new FindersListener());
-      childrenCache.start(StartMode.BUILD_INITIAL_CACHE);
-
-      updateFinders();
-    } catch (Exception e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  @Override
-  public void start() {
-    scanThread = new Thread(new ScanTask(this, env, stopped));
-    scanThread.setName(getClass().getSimpleName() + " " + ScanTask.class.getSimpleName());
-    scanThread.setDaemon(true);
-    scanThread.start();
-  }
-
-  @Override
-  public void stop() {
-    stopped.set(true);
-    try {
-      childrenCache.close();
-    } catch (IOException e1) {
-      log.warn("Failed to close children cache", e1);
-    }
-
-    try {
-      myESNode.close();
-    } catch (IOException e1) {
-      log.warn("Failed to close ephemeral node", e1);
-    }
-
-    scanThread.interrupt();
-    try {
-      scanThread.join();
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  @Override
-  public void failedToProcess(Notification notification, TxResult status) {}
-
-  NotificationProcessor getWorkerQueue() {
-    return notificationProcessor;
-  }
-
-  @VisibleForTesting
-  static boolean shouldProcess(Notification notification, int divisor, int remainder) {
-    byte[] cfcq = NotificationUtil.encodeCol(notification.getColumn());
-    return NotificationHashFilter.accept(ByteUtil.toByteSequence(notification.getRow()),
-        new ArrayByteSequence(cfcq), divisor, remainder);
-  }
-
-  @Override
-  public boolean shouldProcess(Notification notification) {
-    ModulusParams mp = getModulusParams();
-    return shouldProcess(notification, mp.divisor, mp.remainder);
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ModulusParams.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ModulusParams.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ModulusParams.java
deleted file mode 100644
index cbad8b9..0000000
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ModulusParams.java
+++ /dev/null
@@ -1,28 +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.fluo.core.worker.finder.hash;
-
-class ModulusParams {
-  int remainder;
-  int divisor;
-  int update;
-
-  ModulusParams(int r, int d, int u) {
-    this.remainder = r;
-    this.divisor = d;
-    this.update = u;
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ParitionManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ParitionManager.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ParitionManager.java
new file mode 100644
index 0000000..3ebb01e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ParitionManager.java
@@ -0,0 +1,372 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Random;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
+import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
+import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode;
+import org.apache.curator.utils.ZKPaths;
+import org.apache.fluo.accumulo.iterators.NotificationHashFilter;
+import org.apache.fluo.accumulo.util.NotificationUtil;
+import org.apache.fluo.accumulo.util.ZookeeperPath;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.core.impl.Environment;
+import org.apache.fluo.core.impl.FluoConfigurationImpl;
+import org.apache.fluo.core.impl.Notification;
+import org.apache.fluo.core.util.ByteUtil;
+import org.apache.fluo.core.util.FluoThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+/**
+ * This class manages partitioning of notifications across workers coordinating in ZooKeeper.
+ * Workers are divided into groups. Each group is given a subset of the Accumulo table. All workers
+ * in a group scan that subset and use hash partitioning to equally divide notifications.
+ *
+ * <p>
+ * Grouping workers was a compromise between every worker scanning the entire table OR each worker
+ * having a dedicated part of a table. This scheme allows multiple workers to share popular parts of
+ * a table. However, it limits the number of workers that will scan a portion of a table for
+ * notifications. This limitation is important for scaling, even if there are 1,000 workers there
+ * will never be more than 7 to 13 workers scanning a portion of the table.
+ */
+public class ParitionManager {
+
+  private static final Logger log = LoggerFactory.getLogger(ParitionManager.class);
+
+  private final PathChildrenCache childrenCache;
+  private final PersistentEphemeralNode myESNode;
+  private final int groupSize;
+  private long paritionSetTime;
+  private PartitionInfo partitionInfo;
+  private final ScheduledExecutorService schedExecutor;
+
+  private CuratorFramework curator;
+
+  private Environment env;
+
+  private final long minSleepTime;
+  private final long maxSleepTime;
+  private long retrySleepTime;
+
+  private static final long STABILIZE_TIME = TimeUnit.SECONDS.toMillis(60);
+
+  private class FindersListener implements PathChildrenCacheListener {
+
+    @Override
+    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
+      switch (event.getType()) {
+        case CHILD_ADDED:
+        case CHILD_REMOVED:
+        case CHILD_UPDATED:
+          scheduleUpdate();
+          break;
+        default:
+          break;
+      }
+    }
+  }
+
+  static PartitionInfo getGroupInfo(String me, SortedSet<String> children,
+      Collection<TableRange> tablets, int groupSize) {
+
+    int numGroups = Math.max(1, children.size() / groupSize);
+    int[] groupSizes = new int[numGroups];
+    int count = 0;
+    int myGroupId = -1;
+    int myId = -1;
+
+    for (String child : children) {
+      if (child.equals(me)) {
+        myGroupId = count;
+        myId = groupSizes[count];
+      }
+      groupSizes[count]++;
+      count = (count + 1) % numGroups;
+    }
+
+    List<TableRange> rangesCopy = new ArrayList<>(tablets);
+    Collections.sort(rangesCopy);
+
+    // The behavior of Random with a given seed and shuffle are the same across different versions
+    // of java. Both specify the algorithms in their javadoc and are meant to behave the same across
+    // versions. This is important because different workers may be running different versions of
+    // Java, but all workers need to do the same shuffle.
+    //
+    // Did try to use hashing to partition the tablets among groups, but it was slightly uneven. One
+    // group having a 10% more tablets would lead to uneven utilization.
+    Collections.shuffle(rangesCopy, new Random(42));
+
+    List<TableRange> groupsTablets = new ArrayList<>();
+
+    count = 0;
+    for (TableRange tr : rangesCopy) {
+      if (count == myGroupId) {
+        groupsTablets.add(tr);
+      }
+      count = (count + 1) % numGroups;
+    }
+
+    return new PartitionInfo(myId, myGroupId, groupSizes[myGroupId], numGroups, children.size(),
+        groupsTablets);
+  }
+
+  private void updatePartitionInfo() {
+    try {
+      String me = myESNode.getActualPath();
+      while (me == null) {
+        Thread.sleep(100);
+        me = myESNode.getActualPath();
+      }
+      me = ZKPaths.getNodeFromPath(me);
+
+      byte[] zkSplitData = null;
+      SortedSet<String> children = new TreeSet<>();
+      Set<String> groupSizes = new HashSet<>();
+      for (ChildData childData : childrenCache.getCurrentData()) {
+        String node = ZKPaths.getNodeFromPath(childData.getPath());
+        if (node.equals("splits")) {
+          zkSplitData = childData.getData();
+        } else {
+          children.add(node);
+          groupSizes.add(new String(childData.getData(), UTF_8));
+        }
+      }
+
+      if (zkSplitData == null) {
+        log.info("Did not find splits in zookeeper, will retry later.");
+        setPartitionInfo(null); // disable this worker from processing notifications
+        scheduleRetry();
+        return;
+      }
+
+      if (!children.contains(me)) {
+        log.warn("Did not see self (" + me
+            + "), cannot gather tablet and notification partitioning info.");
+        setPartitionInfo(null); // disable this worker from processing notifications
+        scheduleRetry();
+        return;
+      }
+
+      // ensure all workers agree on the group size
+      if (groupSizes.size() != 1 || !groupSizes.contains(groupSize + "")) {
+        log.warn("Group size disagreement " + groupSize + " " + groupSizes
+            + ", cannot gather tablet and notification partitioning info.");
+        setPartitionInfo(null); // disable this worker from processing notifications
+        scheduleRetry();
+        return;
+      }
+
+      List<Bytes> zkSplits = new ArrayList<>();
+      SerializedSplits.deserialize(zkSplits::add, zkSplitData);
+
+      Collection<TableRange> tableRanges = TableRange.toTabletRanges(zkSplits);
+      PartitionInfo newPI = getGroupInfo(me, children, tableRanges, groupSize);
+
+      setPartitionInfo(newPI);
+    } catch (InterruptedException e) {
+      log.debug("Interrupted while gathering tablet and notification partitioning info.", e);
+    } catch (Exception e) {
+      log.warn("Problem gathering tablet and notification partitioning info.", e);
+      setPartitionInfo(null); // disable this worker from processing notifications
+      scheduleRetry();
+    }
+  }
+
+  private synchronized void scheduleRetry() {
+    schedExecutor.schedule(this::updatePartitionInfo, retrySleepTime, TimeUnit.MILLISECONDS);
+    retrySleepTime =
+        Math.min(maxSleepTime,
+            (long) (1.5 * retrySleepTime) + (long) (retrySleepTime * Math.random()));
+  }
+
+  private synchronized void scheduleUpdate() {
+    schedExecutor.schedule(this::updatePartitionInfo, 0, TimeUnit.MILLISECONDS);
+  }
+
+  private class CheckTabletsTask implements Runnable {
+    @Override
+    public void run() {
+      try {
+
+        String me = myESNode.getActualPath();
+        while (me == null) {
+          Thread.sleep(100);
+          me = myESNode.getActualPath();
+        }
+        me = ZKPaths.getNodeFromPath(me);
+
+        String me2 = me;
+        boolean imFirst =
+            childrenCache.getCurrentData().stream().map(ChildData::getPath)
+                .map(ZKPaths::getNodeFromPath).sorted().findFirst().map(s -> s.equals(me2))
+                .orElse(false);
+
+        if (imFirst) {
+
+          ChildData childData = childrenCache.getCurrentData(ZookeeperPath.FINDERS + "/splits");
+          if (childData == null) {
+            byte[] currSplitData = SerializedSplits.serializeTableSplits(env);
+
+            curator.create().forPath(ZookeeperPath.FINDERS + "/splits", currSplitData);
+          } else {
+            HashSet<Bytes> zkSplits = new HashSet<>();
+            SerializedSplits.deserialize(zkSplits::add, childData.getData());
+
+            HashSet<Bytes> currentSplits = new HashSet<>();
+            byte[] currSplitData = SerializedSplits.serializeTableSplits(env);
+            SerializedSplits.deserialize(currentSplits::add, currSplitData);
+
+            if (!currentSplits.equals(zkSplits)) {
+              curator.setData().forPath(ZookeeperPath.FINDERS + "/splits", currSplitData);
+            }
+          }
+        }
+      } catch (InterruptedException e) {
+        log.debug("Interrupted while checking table split points.", e);
+      } catch (Exception e) {
+        log.warn("Failed to checking table split points", e);
+      }
+    }
+  }
+
+  ParitionManager(Environment env, long minSleepTime, long maxSleepTime) {
+    try {
+      this.curator = env.getSharedResources().getCurator();
+      this.env = env;
+
+      this.minSleepTime = minSleepTime;
+      this.maxSleepTime = maxSleepTime;
+      this.retrySleepTime = minSleepTime;
+
+      groupSize =
+          env.getConfiguration().getInt(FluoConfigurationImpl.WORKER_PARTITION_GROUP_SIZE,
+              FluoConfigurationImpl.WORKER_PARTITION_GROUP_SIZE_DEFAULT);
+
+      myESNode =
+          new PersistentEphemeralNode(curator, Mode.EPHEMERAL_SEQUENTIAL, ZookeeperPath.FINDERS
+              + "/f-", ("" + groupSize).getBytes(UTF_8));
+      myESNode.start();
+      myESNode.waitForInitialCreate(1, TimeUnit.MINUTES);
+
+      childrenCache = new PathChildrenCache(curator, ZookeeperPath.FINDERS, true);
+      childrenCache.getListenable().addListener(new FindersListener());
+      childrenCache.start(StartMode.BUILD_INITIAL_CACHE);
+
+      schedExecutor =
+          Executors.newScheduledThreadPool(1,
+              new FluoThreadFactory("Fluo worker partition manager"));
+      schedExecutor.scheduleWithFixedDelay(new CheckTabletsTask(), 0, maxSleepTime,
+          TimeUnit.MILLISECONDS);
+
+      scheduleUpdate();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void setPartitionInfo(PartitionInfo pi) {
+    synchronized (this) {
+      if (!Objects.equals(pi, this.partitionInfo)) {
+        log.debug("Updated finder partition info : " + pi);
+        this.paritionSetTime = System.nanoTime();
+        this.partitionInfo = pi;
+        this.notifyAll();
+      }
+
+      if (pi != null) {
+        retrySleepTime = minSleepTime;
+      }
+    }
+  }
+
+  private long getTimeSincePartitionChange() {
+    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - paritionSetTime);
+  }
+
+  synchronized PartitionInfo waitForPartitionInfo() throws InterruptedException {
+    while (partitionInfo == null
+        || getTimeSincePartitionChange() < Math.min(maxSleepTime, STABILIZE_TIME)) {
+      wait(minSleepTime);
+    }
+
+    return partitionInfo;
+  }
+
+  synchronized PartitionInfo getPartitionInfo() {
+    if (getTimeSincePartitionChange() < Math.min(maxSleepTime, STABILIZE_TIME)) {
+      return null;
+    }
+
+    return partitionInfo;
+  }
+
+  public void stop() {
+    try {
+      myESNode.close();
+    } catch (IOException e) {
+      log.debug("Error closing finder ephemeral node", e);
+    }
+    try {
+      childrenCache.close();
+    } catch (IOException e) {
+      log.debug("Error closing finder children cache", e);
+    }
+
+    schedExecutor.shutdownNow();
+  }
+
+  @VisibleForTesting
+  static boolean shouldProcess(Notification notification, int divisor, int remainder) {
+    byte[] cfcq = NotificationUtil.encodeCol(notification.getColumn());
+    return NotificationHashFilter.accept(ByteUtil.toByteSequence(notification.getRow()),
+        new ArrayByteSequence(cfcq), divisor, remainder);
+  }
+
+  public boolean shouldProcess(Notification notification) {
+    PartitionInfo pi = getPartitionInfo();
+    if (pi == null) {
+      return false;
+    }
+
+    return pi.getMyGroupsRanges().getContaining(notification.getRow()) != null
+        && shouldProcess(notification, pi.getMyGroupSize(), pi.getMyIdInGroup());
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
new file mode 100644
index 0000000..e33b28b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfo.java
@@ -0,0 +1,98 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.List;
+
+class PartitionInfo {
+
+  private final int myGroupId;
+  private final int myIdInGroup;
+  private final int numGroups;
+  private final int myGroupSize;
+  private final int numWorkers;
+  private final RangeSet myGroupsRanges;
+
+  PartitionInfo(int myId, int myGroupId, int myGroupSize, int totalGroups, int totalWorkers,
+      List<TableRange> groupsRanges) {
+    this.myIdInGroup = myId;
+    this.myGroupId = myGroupId;
+    this.myGroupSize = myGroupSize;
+    this.numGroups = totalGroups;
+    this.numWorkers = totalWorkers;
+    this.myGroupsRanges = new RangeSet(groupsRanges);
+  }
+
+  /**
+   * @return The id for the group this worker is in.
+   */
+  public int getMyGroupId() {
+    return myGroupId;
+  }
+
+  /**
+   * @return The id for this worker within its group.
+   */
+  public int getMyIdInGroup() {
+    return myIdInGroup;
+  }
+
+  /**
+   * @return The total number of workers groups there are.
+   */
+  public int getNumGroups() {
+    return numGroups;
+  }
+
+  /**
+   * @return The number of workers in the group this workers is in.
+   */
+  public int getMyGroupSize() {
+    return myGroupSize;
+  }
+
+  /**
+   * @return the total number of workers.
+   */
+  public int getNumWorkers() {
+    return numWorkers;
+  }
+
+  /**
+   * @return the table ranges associated with the group this workers is in.
+   */
+  public RangeSet getMyGroupsRanges() {
+    return myGroupsRanges;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o instanceof PartitionInfo) {
+      PartitionInfo other = (PartitionInfo) o;
+      return other.myGroupId == myGroupId && other.myIdInGroup == myIdInGroup
+          && other.numGroups == numGroups && other.myGroupSize == myGroupSize
+          && other.numWorkers == numWorkers && other.myGroupsRanges.equals(myGroupsRanges);
+    }
+    return false;
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "workers:%d  groups:%d  groupSize:%d  groupId:%d  idInGroup:%d  #tablets:%d", numWorkers,
+        numGroups, myGroupSize, myGroupId, myIdInGroup, myGroupsRanges.size());
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfoChangedException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfoChangedException.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfoChangedException.java
new file mode 100644
index 0000000..35c3f9b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionInfoChangedException.java
@@ -0,0 +1,20 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+public class PartitionInfoChangedException extends RuntimeException {
+  private static final long serialVersionUID = 1L;
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionNotificationFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionNotificationFinder.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionNotificationFinder.java
new file mode 100644
index 0000000..c940850
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/PartitionNotificationFinder.java
@@ -0,0 +1,84 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.fluo.core.impl.Environment;
+import org.apache.fluo.core.impl.FluoConfigurationImpl;
+import org.apache.fluo.core.impl.Notification;
+import org.apache.fluo.core.worker.NotificationFinder;
+import org.apache.fluo.core.worker.NotificationProcessor;
+import org.apache.fluo.core.worker.TxResult;
+
+public class PartitionNotificationFinder implements NotificationFinder {
+
+  private ParitionManager paritionManager;
+  private Thread scanThread;
+  private NotificationProcessor processor;
+  private Environment env;
+  private AtomicBoolean stopped;
+
+  @Override
+  public void init(Environment env, NotificationProcessor processor) {
+    this.processor = processor;
+    this.env = env;
+    this.stopped = new AtomicBoolean(false);
+
+  }
+
+  @Override
+  public void start() {
+    long minSleepTime =
+        env.getConfiguration().getInt(FluoConfigurationImpl.NTFY_FINDER_MIN_SLEEP_TIME_PROP,
+            FluoConfigurationImpl.NTFY_FINDER_MIN_SLEEP_TIME_DEFAULT);
+    long maxSleepTime =
+        env.getConfiguration().getInt(FluoConfigurationImpl.NTFY_FINDER_MAX_SLEEP_TIME_PROP,
+            FluoConfigurationImpl.NTFY_FINDER_MAX_SLEEP_TIME_DEFAULT);
+
+    paritionManager = new ParitionManager(env, minSleepTime, maxSleepTime);
+
+    scanThread =
+        new Thread(new ScanTask(this, processor, paritionManager, env, stopped, minSleepTime,
+            maxSleepTime));
+    scanThread.setName(getClass().getSimpleName() + " " + ScanTask.class.getSimpleName());
+    scanThread.setDaemon(true);
+    scanThread.start();
+  }
+
+  @Override
+  public void stop() {
+    stopped.set(true);
+
+    scanThread.interrupt();
+    try {
+      scanThread.join();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+
+    paritionManager.stop();
+  }
+
+  @Override
+  public boolean shouldProcess(Notification notification) {
+    return paritionManager.shouldProcess(notification);
+  }
+
+  @Override
+  public void failedToProcess(Notification notification, TxResult status) {}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
new file mode 100644
index 0000000..40dc5e4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/RangeSet.java
@@ -0,0 +1,88 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.TreeMap;
+import java.util.function.Consumer;
+
+import org.apache.fluo.api.data.Bytes;
+
+public class RangeSet {
+  private TreeMap<Bytes, TableRange> tmap;
+  private TableRange lastRange;
+
+  public RangeSet(List<TableRange> ranges) {
+    tmap = new TreeMap<>();
+
+    for (TableRange tablet : ranges) {
+      if (tablet.getEndRow() == null) {
+        lastRange = tablet;
+      } else {
+        tmap.put(tablet.getEndRow(), tablet);
+      }
+    }
+  }
+
+  public TableRange getContaining(Bytes row) {
+    Entry<Bytes, TableRange> entry = tmap.ceilingEntry(row);
+    if (entry != null) {
+      if (entry.getValue().contains(row)) {
+        return entry.getValue();
+      }
+    } else if (lastRange != null) {
+      if (lastRange.contains(row)) {
+        return lastRange;
+      }
+    }
+
+    return null;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o instanceof RangeSet) {
+      RangeSet ots = (RangeSet) o;
+
+      if (tmap.size() != ots.tmap.size()) {
+        return false;
+      }
+
+      for (Entry<Bytes, TableRange> entry : tmap.entrySet()) {
+        TableRange otr = ots.tmap.get(entry.getKey());
+        if (!Objects.equals(entry.getValue(), otr)) {
+          return false;
+        }
+      }
+
+      return Objects.equals(lastRange, ots.lastRange);
+    }
+    return false;
+  }
+
+  public int size() {
+    return tmap.size() + (lastRange == null ? 0 : 1);
+  }
+
+  public void forEach(Consumer<TableRange> trc) {
+    if (lastRange != null) {
+      trc.accept(lastRange);
+    }
+    tmap.values().forEach(trc);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ScanTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ScanTask.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ScanTask.java
index e0fa791..1732a29 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ScanTask.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/ScanTask.java
@@ -17,11 +17,14 @@ package org.apache.fluo.core.worker.finder.hash;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.Scanner;
@@ -31,12 +34,11 @@ import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.fluo.accumulo.iterators.NotificationHashFilter;
 import org.apache.fluo.core.impl.Environment;
-import org.apache.fluo.core.impl.FluoConfigurationImpl;
 import org.apache.fluo.core.impl.Notification;
 import org.apache.fluo.core.util.UtilWaitThread;
-import org.apache.fluo.core.worker.TabletInfoCache;
-import org.apache.fluo.core.worker.TabletInfoCache.TabletInfo;
-import org.apache.fluo.core.worker.finder.hash.HashNotificationFinder.ModParamsChangedException;
+import org.apache.fluo.core.worker.NotificationFinder;
+import org.apache.fluo.core.worker.NotificationProcessor;
+import org.apache.fluo.core.worker.NotificationProcessor.Session;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,91 +46,101 @@ public class ScanTask implements Runnable {
 
   private static final Logger log = LoggerFactory.getLogger(ScanTask.class);
 
-  private final HashNotificationFinder hwf;
+  private final NotificationFinder finder;
+  private final ParitionManager partitionManager;
+  private final NotificationProcessor proccessor;
   private final Random rand = new Random();
   private final AtomicBoolean stopped;
-  private final TabletInfoCache<TabletData, Supplier<TabletData>> tabletInfoCache;
+  private final Map<TableRange, TabletData> rangeData;
   private final Environment env;
 
-  static long STABILIZE_TIME = 10 * 1000;
-
   private long minSleepTime;
   private long maxSleepTime;
 
-  ScanTask(HashNotificationFinder hashWorkFinder, Environment env, AtomicBoolean stopped) {
-    this.hwf = hashWorkFinder;
-    this.tabletInfoCache = new TabletInfoCache<>(env, new Supplier<TabletData>() {
-      @Override
-      public TabletData get() {
-        return new TabletData();
-      }
-    });
+  ScanTask(NotificationFinder finder, NotificationProcessor proccessor,
+      ParitionManager partitionManager, Environment env, AtomicBoolean stopped, long minSleepTime,
+      long maxSleepTime) {
+    this.finder = finder;
+    this.rangeData = new HashMap<>();
+
     this.env = env;
     this.stopped = stopped;
 
-    minSleepTime =
-        env.getConfiguration().getInt(FluoConfigurationImpl.MIN_SLEEP_TIME_PROP,
-            FluoConfigurationImpl.MIN_SLEEP_TIME_DEFAULT);
-    maxSleepTime =
-        env.getConfiguration().getInt(FluoConfigurationImpl.MAX_SLEEP_TIME_PROP,
-            FluoConfigurationImpl.MAX_SLEEP_TIME_DEFAULT);
+    this.proccessor = proccessor;
+    this.partitionManager = partitionManager;
+
+    this.minSleepTime = minSleepTime;
+    this.maxSleepTime = maxSleepTime;
   }
 
   @Override
   public void run() {
 
-    int qSize = hwf.getWorkerQueue().size();
+
+    List<TableRange> ranges = new ArrayList<>();
+    Set<TableRange> rangeSet = new HashSet<>();
+
+    int qSize = proccessor.size();
 
     while (!stopped.get()) {
       try {
+        PartitionInfo partition = partitionManager.waitForPartitionInfo();
 
-        while (hwf.getWorkerQueue().size() > qSize / 2 && !stopped.get()) {
+        while (proccessor.size() > qSize / 2 && !stopped.get()) {
           UtilWaitThread.sleep(50, stopped);
         }
 
-        // break scan work into a lot of ranges that are randomly ordered. This has a few benefits.
-        // Ensures different workers are scanning different tablets.
-        // Allows checking local state more frequently in the case where work is not present in many
-        // tablets. Allows less frequent scanning of tablets that are
-        // usually empty.
-        List<TabletInfo<TabletData>> tablets = new ArrayList<>(tabletInfoCache.getTablets());
-        Collections.shuffle(tablets, rand);
+        ranges.clear();
+        rangeSet.clear();
+        partition.getMyGroupsRanges().forEach(t -> {
+          ranges.add(t);
+          rangeSet.add(t);
+        });
+        Collections.shuffle(ranges, rand);
+        rangeData.keySet().retainAll(rangeSet);
 
         long minRetryTime = maxSleepTime + System.currentTimeMillis();
         int notifications = 0;
         int tabletsScanned = 0;
         try {
-          for (TabletInfo<TabletData> tabletInfo : tablets) {
-            if (System.currentTimeMillis() >= tabletInfo.getData().retryTime) {
+          for (TableRange tabletRange : ranges) {
+            TabletData tabletData = rangeData.computeIfAbsent(tabletRange, tr -> new TabletData());
+            if (System.currentTimeMillis() >= tabletData.retryTime) {
               int count = 0;
-              ModulusParams modParams = hwf.getModulusParams();
-              if (modParams != null) {
-                // notifications could have been asynchronously queued for deletion. Let that happen
-                // 1st before scanning
-                env.getSharedResources().getBatchWriter().waitForAsyncFlush();
-                count = scan(modParams, tabletInfo.getRange());
-                tabletsScanned++;
+              PartitionInfo pi = partitionManager.getPartitionInfo();
+              if (partition.equals(pi)) {
+                try (Session session =
+                    proccessor.beginAddingNotifications(rc -> tabletRange.contains(rc.getRow()))) {
+                  // notifications could have been asynchronously queued for deletion. Let that
+                  // happen
+                  // 1st before scanning
+                  env.getSharedResources().getBatchWriter().waitForAsyncFlush();
+
+                  count = scan(session, partition, tabletRange.getRange());
+                  tabletsScanned++;
+                }
+              } else {
+                break;
               }
-              tabletInfo.getData().updateScanCount(count, maxSleepTime);
+              tabletData.updateScanCount(count, maxSleepTime);
               notifications += count;
               if (stopped.get()) {
                 break;
               }
             }
 
-            minRetryTime = Math.min(tabletInfo.getData().retryTime, minRetryTime);
+            minRetryTime = Math.min(tabletData.retryTime, minRetryTime);
           }
-        } catch (ModParamsChangedException mpce) {
-          hwf.getWorkerQueue().clear();
-          waitForFindersToStabilize();
+        } catch (PartitionInfoChangedException mpce) {
+          // nothing to do
         }
 
         long sleepTime = Math.max(minSleepTime, minRetryTime - System.currentTimeMillis());
 
-        qSize = hwf.getWorkerQueue().size();
+        qSize = proccessor.size();
 
         log.debug("Scanned {} of {} tablets, added {} new notifications (total queued {})",
-            tabletsScanned, tablets.size(), notifications, qSize);
+            tabletsScanned, ranges.size(), notifications, qSize);
 
         if (!stopped.get()) {
           UtilWaitThread.sleep(sleepTime, stopped);
@@ -141,7 +153,6 @@ public class ScanTask implements Runnable {
           log.error("Error while looking for notifications", e);
         }
       }
-
     }
   }
 
@@ -157,7 +168,7 @@ public class ScanTask implements Runnable {
     return wasInt;
   }
 
-  private int scan(ModulusParams lmp, Range range) throws TableNotFoundException {
+  private int scan(Session session, PartitionInfo pi, Range range) throws TableNotFoundException {
     Scanner scanner = env.getConnector().createScanner(env.getTable(), env.getAuthorizations());
 
     scanner.setRange(range);
@@ -165,38 +176,24 @@ public class ScanTask implements Runnable {
     Notification.configureScanner(scanner);
 
     IteratorSetting iterCfg = new IteratorSetting(30, "nhf", NotificationHashFilter.class);
-    NotificationHashFilter.setModulusParams(iterCfg, lmp.divisor, lmp.remainder);
+    NotificationHashFilter.setModulusParams(iterCfg, pi.getMyGroupSize(), pi.getMyIdInGroup());
     scanner.addScanIterator(iterCfg);
 
     int count = 0;
 
     for (Entry<Key, Value> entry : scanner) {
-      if (lmp.update != hwf.getModulusParams().update) {
-        throw new HashNotificationFinder.ModParamsChangedException();
+      if (!pi.equals(partitionManager.getPartitionInfo())) {
+        throw new PartitionInfoChangedException();
       }
 
       if (stopped.get()) {
         return count;
       }
 
-      if (hwf.getWorkerQueue().addNotification(hwf, Notification.from(entry.getKey()))) {
+      if (session.addNotification(finder, Notification.from(entry.getKey()))) {
         count++;
       }
     }
     return count;
   }
-
-  private void waitForFindersToStabilize() {
-    ModulusParams lmp = hwf.getModulusParams();
-    long startTime = System.currentTimeMillis();
-
-    while (System.currentTimeMillis() - startTime < STABILIZE_TIME) {
-      UtilWaitThread.sleep(500, stopped);
-      ModulusParams lmp2 = hwf.getModulusParams();
-      if (lmp.update != lmp2.update) {
-        startTime = System.currentTimeMillis();
-        lmp = lmp2;
-      }
-    }
-  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/SerializedSplits.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/SerializedSplits.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/SerializedSplits.java
new file mode 100644
index 0000000..53ad7ba
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/SerializedSplits.java
@@ -0,0 +1,119 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.io.BufferedOutputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Bytes.BytesBuilder;
+import org.apache.fluo.core.impl.Environment;
+import org.apache.fluo.core.util.ByteUtil;
+
+public class SerializedSplits {
+
+  static final int MAX_SIZE = 1 << 18;
+
+  public static void deserialize(Consumer<Bytes> splitConsumer, byte[] serializedSplits) {
+    try {
+      ByteArrayInputStream bais = new ByteArrayInputStream(serializedSplits);
+      GZIPInputStream gzis = new GZIPInputStream(bais);
+      DataInputStream dis = new DataInputStream(gzis);
+
+      int numSplits = dis.readInt();
+
+      BytesBuilder builder = Bytes.builder();
+
+      for (int i = 0; i < numSplits; i++) {
+        int len = dis.readInt();
+        builder.setLength(0);
+        builder.append(dis, len);
+        splitConsumer.accept(builder.toBytes());
+      }
+
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  private static byte[] serializeInternal(List<Bytes> splits) throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    GZIPOutputStream gzOut = new GZIPOutputStream(baos);
+    BufferedOutputStream bos = new BufferedOutputStream(gzOut, 1 << 16);
+    DataOutputStream dos = new DataOutputStream(bos);
+
+    dos.writeInt(splits.size());
+    for (Bytes split : splits) {
+      dos.writeInt(split.length());
+      split.writeTo(dos);
+    }
+
+    dos.close();
+
+    return baos.toByteArray();
+  }
+
+  public static byte[] serialize(Collection<Bytes> splits) {
+    List<Bytes> splitsCopy = new ArrayList<>(splits);
+    Collections.sort(splitsCopy);
+
+    try {
+      byte[] serialized = serializeInternal(splitsCopy);
+
+      while (serialized.length > MAX_SIZE) {
+        List<Bytes> splitsCopy2 = new ArrayList<>(splitsCopy.size() / 2 + 1);
+        for (int i = 0; i < splitsCopy.size(); i += 2) {
+          splitsCopy2.add(splitsCopy.get(i));
+        }
+
+        splitsCopy = splitsCopy2;
+        serialized = serializeInternal(splitsCopy);
+      }
+
+      return serialized;
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  public static byte[] serializeTableSplits(Environment env) {
+    List<Bytes> splits;
+    try {
+      splits =
+          env.getConnector().tableOperations().listSplits(env.getTable()).stream()
+              .map(ByteUtil::toBytes).collect(Collectors.toList());
+    } catch (TableNotFoundException | AccumuloSecurityException | AccumuloException e) {
+      throw new RuntimeException(e);
+    }
+    return serialize(splits);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/TableRange.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/TableRange.java b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/TableRange.java
new file mode 100644
index 0000000..30a97f4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/finder/hash/TableRange.java
@@ -0,0 +1,113 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.accumulo.core.data.Range;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.core.util.ByteUtil;
+import org.apache.hadoop.io.Text;
+
+import static java.util.stream.Collectors.toList;
+
+public class TableRange implements Comparable<TableRange> {
+  private final Bytes prevEndRow;
+  private final Bytes endRow;
+  private final int hc;
+
+  public TableRange(Bytes per, Bytes er) {
+    this.prevEndRow = per;
+    this.endRow = er;
+    this.hc = Objects.hash(this.prevEndRow, this.endRow);
+  }
+
+  @Override
+  public int hashCode() {
+    return hc;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o instanceof TableRange) {
+      TableRange ot = (TableRange) o;
+      return Objects.equals(prevEndRow, ot.prevEndRow) && Objects.equals(endRow, ot.endRow);
+    }
+
+    return false;
+  }
+
+  public Bytes getPrevEndRow() {
+    return prevEndRow;
+  }
+
+  public Bytes getEndRow() {
+    return endRow;
+  }
+
+  public boolean contains(Bytes row) {
+    return (prevEndRow == null || row.compareTo(prevEndRow) > 0)
+        && (endRow == null || row.compareTo(endRow) <= 0);
+  }
+
+  @Override
+  public String toString() {
+    return getPrevEndRow() + " " + getEndRow();
+  }
+
+
+  public static Collection<TableRange> toTabletRanges(Collection<Bytes> rows) {
+    List<Bytes> sortedRows = rows.stream().sorted().collect(toList());
+    List<TableRange> tablets = new ArrayList<>(sortedRows.size() + 1);
+    for (int i = 0; i < sortedRows.size(); i++) {
+      tablets.add(new TableRange(i == 0 ? null : sortedRows.get(i - 1), sortedRows.get(i)));
+    }
+
+    tablets.add(new TableRange(sortedRows.size() == 0 ? null
+        : sortedRows.get(sortedRows.size() - 1), null));
+    return tablets;
+  }
+
+
+
+  public Range getRange() {
+    Text tper = Optional.ofNullable(prevEndRow).map(ByteUtil::toText).orElse(null);
+    Text ter = Optional.ofNullable(endRow).map(ByteUtil::toText).orElse(null);
+    return new Range(tper, false, ter, true);
+  }
+
+  @Override
+  public int compareTo(TableRange o) {
+    if (Objects.equals(getEndRow(), o.getEndRow())) {
+      // this will catch case of both null
+      return 0;
+    }
+
+    if (getEndRow() == null) {
+      return 1;
+    }
+
+    if (o.getEndRow() == null) {
+      return -1;
+    }
+
+    return getEndRow().compareTo(o.getEndRow());
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/HashTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/HashTest.java b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/HashTest.java
index 6790e54..b3496dd 100644
--- a/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/HashTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/HashTest.java
@@ -63,7 +63,7 @@ public class HashTest {
     byte[] cfcq = NotificationUtil.encodeCol(col);
     Key k = new Key(row, ColumnConstants.NOTIFY_CF.toArray(), cfcq, new byte[0], 6);
     boolean accept = NotificationHashFilter.accept(k, 7, 3);
-    Assert.assertEquals(accept, HashNotificationFinder.shouldProcess(Notification.from(k), 7, 3));
+    Assert.assertEquals(accept, ParitionManager.shouldProcess(Notification.from(k), 7, 3));
     return accept;
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/PartitionManagerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/PartitionManagerTest.java b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/PartitionManagerTest.java
new file mode 100644
index 0000000..8a02edc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/PartitionManagerTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.IntSummaryStatistics;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.function.IntFunction;
+import java.util.stream.IntStream;
+
+import org.apache.fluo.api.data.Bytes;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static java.util.stream.Collectors.toList;
+
+public class PartitionManagerTest {
+
+  @Test
+  public void testGrouping() {
+    IntFunction<String> nff = i -> String.format("f-%04d", i);
+
+    for (int numSplits : new int[] {1, 10, 100, 1000}) {
+      for (int numWorkers : new int[] {1, 5, 10, 11, 100}) {
+        for (int groupSize : new int[] {1, 2, 3, 5, 7, 13, 17, 19, 43, 73, 97}) {
+          int expectedGroups = Math.max(1, numWorkers / groupSize);
+          int maxGroupSize =
+              Math.min(numWorkers,
+                  groupSize + (int) Math.ceil((numWorkers % groupSize) / (double) expectedGroups));
+
+          TreeSet<String> children = new TreeSet<>();
+
+          IntStream.range(0, numWorkers).mapToObj(nff).forEach(children::add);
+
+          Collection<Bytes> rows =
+              IntStream.iterate(0, i -> i + 1000).limit(numSplits)
+                  .mapToObj(i -> String.format("r%06d", i)).map(Bytes::of).collect(toList());
+          Collection<TableRange> tablets = TableRange.toTabletRanges(rows);
+
+          Set<String> idCombos = new HashSet<>();
+          Map<Integer, RangeSet> groupTablets = new HashMap<>();
+
+          for (int i = 0; i < numWorkers; i++) {
+            String me = nff.apply(i);
+            PartitionInfo pi = ParitionManager.getGroupInfo(me, children, tablets, groupSize);
+            Assert.assertEquals(expectedGroups, pi.getNumGroups());
+            Assert.assertTrue(pi.getMyGroupSize() >= Math.min(numWorkers, groupSize)
+                && pi.getMyGroupSize() <= maxGroupSize);
+            Assert.assertEquals(numWorkers, pi.getNumWorkers());
+            Assert.assertTrue(pi.getMyIdInGroup() >= 0 && pi.getMyIdInGroup() < maxGroupSize);
+            Assert.assertTrue(pi.getMyGroupId() >= 0 && pi.getMyGroupId() < expectedGroups);
+
+            Assert.assertFalse(idCombos.contains(pi.getMyGroupId() + ":" + pi.getMyIdInGroup()));
+            idCombos.add(pi.getMyGroupId() + ":" + pi.getMyIdInGroup());
+
+            if (!groupTablets.containsKey(pi.getMyGroupId())) {
+              groupTablets.put(pi.getMyGroupId(), pi.getMyGroupsRanges());
+            } else {
+              Assert.assertEquals(groupTablets.get(pi.getMyGroupId()), pi.getMyGroupsRanges());
+            }
+          }
+
+          Assert.assertEquals(numWorkers, idCombos.size());
+
+          // check that the tablets for each group are disjoint and that the union of the tablets
+          // for each group has all tablets
+          HashSet<TableRange> allTabletsFromGroups = new HashSet<>();
+
+          for (RangeSet tabletSet : groupTablets.values()) {
+            tabletSet.forEach(tr -> {
+              Assert.assertFalse(allTabletsFromGroups.contains(tr));
+              allTabletsFromGroups.add(tr);
+            });
+          }
+
+          Assert.assertEquals(new HashSet<>(tablets), allTabletsFromGroups);
+
+          // check that all groups have about the same number of tablets
+          IntSummaryStatistics summaryStats =
+              groupTablets.values().stream().mapToInt(RangeSet::size).summaryStatistics();
+          Assert.assertTrue(summaryStats.getMax() - summaryStats.getMin() < 2);
+        }
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/SerializedSplitsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/SerializedSplitsTest.java b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/SerializedSplitsTest.java
new file mode 100644
index 0000000..13aa941
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/SerializedSplitsTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import org.apache.fluo.api.data.Bytes;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
+
+public class SerializedSplitsTest {
+  @Test
+  public void testLotsOfSplits() {
+    List<Bytes> splits =
+        IntStream.iterate(0, i -> i + 13).limit(1_000_000).mapToObj(i -> String.format("%08x", i))
+            .map(Bytes::of).collect(toList());
+
+    byte[] data = SerializedSplits.serialize(splits);
+    Assert.assertTrue(data.length <= SerializedSplits.MAX_SIZE);
+
+    List<Bytes> splits2 = new ArrayList<>();
+    SerializedSplits.deserialize(splits2::add, data);
+
+    Assert.assertTrue(splits2.size() > 10_000);
+    Assert.assertTrue(splits2.size() < splits.size());
+
+    int expectedDiff =
+        Integer.parseInt(splits2.get(1).toString(), 16)
+            - Integer.parseInt(splits2.get(0).toString(), 16);
+    Assert.assertTrue(expectedDiff > 13);
+    Assert.assertTrue(expectedDiff % 13 == 0);
+    // check that splits are evenly spaced
+    for (int i = 1; i < splits2.size(); i++) {
+      int sp1 = Integer.parseInt(splits2.get(i - 1).toString(), 16);
+      int sp2 = Integer.parseInt(splits2.get(i).toString(), 16);
+      Assert.assertEquals(expectedDiff, sp2 - sp1);
+      Assert.assertEquals(0, sp1 % 13);
+      Assert.assertEquals(0, sp2 % 13);
+    }
+
+    Assert.assertTrue(splits.get(0).compareTo(splits2.get(0)) <= 0);
+    Assert
+        .assertTrue(splits.get(splits.size() - 1).compareTo(splits2.get(splits2.size() - 1)) >= 0);
+  }
+
+  @Test
+  public void testSimple() {
+    Set<Bytes> splits =
+        IntStream.iterate(0, i -> i + 13).limit(1_000).mapToObj(i -> String.format("%08x", i))
+            .map(Bytes::of).collect(toSet());
+
+    byte[] data = SerializedSplits.serialize(splits);
+
+    HashSet<Bytes> splits2 = new HashSet<>();
+    SerializedSplits.deserialize(splits2::add, data);
+
+    Assert.assertEquals(splits, splits2);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/TableRangeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/TableRangeTest.java b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/TableRangeTest.java
new file mode 100644
index 0000000..c73044a
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/core/worker/finder/hash/TableRangeTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.fluo.core.worker.finder.hash;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import org.apache.accumulo.core.data.Range;
+import org.apache.fluo.api.data.Bytes;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TableRangeTest {
+  @Test
+  public void testBasic() {
+    TableRange tr1 = new TableRange(null, null);
+
+    Assert.assertTrue(tr1.contains(Bytes.of("a")));
+    Assert.assertTrue(tr1.contains(Bytes.of("z")));
+    Assert.assertNull(tr1.getEndRow());
+    Assert.assertNull(tr1.getPrevEndRow());
+
+    TableRange tr2 = new TableRange(null, Bytes.of("ma"));
+    Assert.assertTrue(tr2.contains(Bytes.of("a")));
+    Assert.assertTrue(tr2.contains(Bytes.of("ma")));
+    Assert.assertFalse(tr2.contains(Bytes.of("maa")));
+    Assert.assertFalse(tr2.contains(Bytes.of("z")));
+    Assert.assertNull(tr2.getPrevEndRow());
+    Assert.assertEquals(Bytes.of("ma"), tr2.getEndRow());
+
+    TableRange tr3 = new TableRange(Bytes.of("la"), null);
+    Assert.assertFalse(tr3.contains(Bytes.of("a")));
+    Assert.assertFalse(tr3.contains(Bytes.of("la")));
+    Assert.assertTrue(tr3.contains(Bytes.of("laa")));
+    Assert.assertTrue(tr3.contains(Bytes.of("z")));
+    Assert.assertEquals(Bytes.of("la"), tr3.getPrevEndRow());
+    Assert.assertNull(tr3.getEndRow());
+
+    TableRange tr4 = new TableRange(Bytes.of("la"), Bytes.of("ma"));
+    Assert.assertFalse(tr4.contains(Bytes.of("a")));
+    Assert.assertFalse(tr4.contains(Bytes.of("la")));
+    Assert.assertTrue(tr4.contains(Bytes.of("laa")));
+    Assert.assertTrue(tr4.contains(Bytes.of("ma")));
+    Assert.assertFalse(tr4.contains(Bytes.of("maa")));
+    Assert.assertFalse(tr4.contains(Bytes.of("z")));
+    Assert.assertEquals(Bytes.of("la"), tr4.getPrevEndRow());
+    Assert.assertEquals(Bytes.of("ma"), tr4.getEndRow());
+  }
+
+  @Test
+  public void testMultiple() {
+
+    Bytes sp1 = Bytes.of("e1");
+    Bytes sp2 = Bytes.of("m1");
+    Bytes sp3 = Bytes.of("r1");
+
+    Collection<TableRange> trc1 =
+        new HashSet<>(TableRange.toTabletRanges(Arrays.asList(sp2, sp3, sp1)));
+
+    Assert.assertEquals(4, trc1.size());
+    Assert.assertTrue(trc1.contains(new TableRange(null, sp1)));
+    Assert.assertTrue(trc1.contains(new TableRange(sp1, sp2)));
+    Assert.assertTrue(trc1.contains(new TableRange(sp2, sp3)));
+    Assert.assertTrue(trc1.contains(new TableRange(sp3, null)));
+
+    Collection<TableRange> trc2 = new HashSet<>(TableRange.toTabletRanges(Collections.emptyList()));
+    Assert.assertEquals(1, trc2.size());
+    Assert.assertTrue(trc2.contains(new TableRange(null, null)));
+  }
+
+  @Test
+  public void testCompare() {
+
+    Bytes sp1 = Bytes.of("e1");
+    Bytes sp2 = Bytes.of("m1");
+
+    TableRange tr1 = new TableRange(null, sp1);
+    TableRange tr2 = new TableRange(sp1, sp2);
+    TableRange tr3 = new TableRange(sp2, null);
+
+    Assert.assertTrue(tr1.compareTo(tr2) < 0);
+    Assert.assertTrue(tr2.compareTo(tr1) > 0);
+
+    Assert.assertTrue(tr2.compareTo(tr3) < 0);
+    Assert.assertTrue(tr3.compareTo(tr2) > 0);
+
+    Assert.assertTrue(tr1.compareTo(tr3) < 0);
+    Assert.assertTrue(tr3.compareTo(tr1) > 0);
+
+    Assert.assertTrue(tr1.compareTo(tr1) == 0);
+    Assert.assertTrue(tr2.compareTo(tr2) == 0);
+    Assert.assertTrue(tr3.compareTo(tr3) == 0);
+
+    Assert.assertTrue(tr1.compareTo(new TableRange(null, sp1)) == 0);
+    Assert.assertTrue(tr2.compareTo(new TableRange(sp1, sp2)) == 0);
+    Assert.assertTrue(tr3.compareTo(new TableRange(sp2, null)) == 0);
+
+    Assert.assertTrue(new TableRange(null, null).compareTo(new TableRange(null, null)) == 0);
+  }
+
+  @Test
+  public void testToRange() {
+    for (String prev : new String[] {null, "foo"}) {
+      for (String end : new String[] {null, "zoo"}) {
+        Assert.assertEquals(new Range(prev, false, end, true), new TableRange(prev == null ? null
+            : Bytes.of(prev), end == null ? null : Bytes.of(end)).getRange());
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
----------------------------------------------------------------------
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
index 9db0ce4..66791b5 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
@@ -28,7 +28,7 @@ import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.TransactionImpl.CommitData;
 import org.apache.fluo.core.observer.Observers;
 import org.apache.fluo.core.worker.NotificationFinder;
-import org.apache.fluo.core.worker.finder.hash.HashNotificationFinder;
+import org.apache.fluo.core.worker.finder.hash.PartitionNotificationFinder;
 import org.apache.fluo.integration.ITBaseMini;
 import org.apache.fluo.integration.TestTransaction;
 import org.apache.fluo.mini.MiniFluoImpl;
@@ -171,11 +171,11 @@ public class WorkerIT extends ITBaseMini {
 
     try (Environment env = new Environment(config)) {
 
-      NotificationFinder nf1 = new HashNotificationFinder();
+      NotificationFinder nf1 = new PartitionNotificationFinder();
       nf1.init(env, ((MiniFluoImpl) miniFluo).getNotificationProcessor());
       nf1.start();
 
-      NotificationFinder nf2 = new HashNotificationFinder();
+      NotificationFinder nf2 = new PartitionNotificationFinder();
       nf2.init(env, ((MiniFluoImpl) miniFluo).getNotificationProcessor());
       nf2.start();
 

http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/3da76dcf/modules/mini/src/main/java/org/apache/fluo/mini/MiniFluoImpl.java
----------------------------------------------------------------------
diff --git a/modules/mini/src/main/java/org/apache/fluo/mini/MiniFluoImpl.java b/modules/mini/src/main/java/org/apache/fluo/mini/MiniFluoImpl.java
index 4e798ab..6d1ce5d 100644
--- a/modules/mini/src/main/java/org/apache/fluo/mini/MiniFluoImpl.java
+++ b/modules/mini/src/main/java/org/apache/fluo/mini/MiniFluoImpl.java
@@ -84,8 +84,8 @@ public class MiniFluoImpl implements MiniFluo {
         startMiniAccumulo();
       }
 
-      config.setProperty(FluoConfigurationImpl.MIN_SLEEP_TIME_PROP, 50);
-      config.setProperty(FluoConfigurationImpl.MAX_SLEEP_TIME_PROP, 100);
+      config.setProperty(FluoConfigurationImpl.NTFY_FINDER_MIN_SLEEP_TIME_PROP, 50);
+      config.setProperty(FluoConfigurationImpl.NTFY_FINDER_MAX_SLEEP_TIME_PROP, 100);
 
       env = new Environment(config);
 



Mime
View raw message