diff --git a/.gitignore b/.gitignore
index 9ad90d3c..e84b067d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,9 @@ lib_managed/
 src_managed/
 project/boot/
 project/plugins/project
+out/
+build/
+.gradle
+*.iml
+*.ipr
+*.iws
diff --git a/README.md b/README.md
index 7af588db..47f1f364 100644
--- a/README.md
+++ b/README.md
@@ -269,3 +269,4 @@ If you are building a partitioned cluster then you will want to use the `Partiti
 * maxConnectionsPerNode - the maximum number of open connections to a node. The total number of connections that can be opened by a network client is maxConnectionsPerNode * number of nodes
 * staleRequestTimeoutMins - the number of minutes to keep a request that is waiting for a response
 * staleRequestCleanupFrequenceMins - the frequency to clean up stale requests
+
diff --git a/build.gradle b/build.gradle
new file mode 100644
index 00000000..ed676889
--- /dev/null
+++ b/build.gradle
@@ -0,0 +1,27 @@
+project.ext.isDefaultEnvironment = !project.hasProperty('overrideBuildEnvironment')
+
+File getEnvironmentScript()
+{
+  final File env = file(isDefaultEnvironment ? 'defaultEnvironment.gradle' : project.overrideBuildEnvironment)
+  assert env.isFile() : "The environment script [$env] does not exists or is not a file."
+  return env
+}
+
+apply from: environmentScript
+
+project.ext.externalDependency = [
+  'zookeeper':'org.apache.zookeeper:zookeeper:3.3.0',
+  'protobuf':'com.google.protobuf:protobuf-java:2.4.0a',
+  'log4j':'log4j:log4j:1.2.16',
+  'netty':'io.netty:netty:3.5.11.Final',
+  'slf4jApi':'org.slf4j:slf4j-api:1.5.6',
+  'slf4jLog4j':'org.slf4j:slf4j-log4j12:1.5.6',
+  'specs':'org.scala-tools.testing:specs_2.8.1:1.6.8',
+  'mockitoAll':'org.mockito:mockito-all:1.8.4',
+  'cglib':'cglib:cglib:2.1_3',
+  'objenesis':'org.objenesis:objenesis:1.2',
+  'scalaCompiler': 'org.scala-lang:scala-compiler:2.8.1',
+  'scalaLibrary': 'org.scala-lang:scala-library:2.8.1',
+  'scalatest': 'org.scalatest:scalatest:1.2'
+];
+
diff --git a/cluster/build.gradle b/cluster/build.gradle
new file mode 100644
index 00000000..dffd140c
--- /dev/null
+++ b/cluster/build.gradle
@@ -0,0 +1,18 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile externalDependency.scalaLibrary 
+  compile externalDependency.zookeeper
+  compile externalDependency.protobuf
+  compile externalDependency.log4j
+
+  testCompile externalDependency.specs
+  testCompile externalDependency.mockitoAll
+  testCompile externalDependency.cglib
+  testCompile externalDependency.objenesis
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+
diff --git a/cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala b/cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala
index e7c8474f..8371c07d 100644
--- a/cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala
+++ b/cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala
@@ -37,6 +37,8 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
     //synchronization is not needed between this method
     case class NodeChildrenChanged(path: String) extends ZooKeeperMessage
     case class NodeDataChanged(path: String) extends ZooKeeperMessage
+    case class NodeDeleted(path: String) extends ZooKeeperMessage
+    case class NodeCreated(path: String) extends ZooKeeperMessage
   }
 
   class ZooKeeperClusterManager(connectString: String, sessionTimeout: Int, serviceName: String)
@@ -90,6 +92,19 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
             handleCapabilityMemberChanged(path)
           }
 
+          case NodeDeleted(path) => if (path.equals(MEMBERSHIP_NODE) || path.equals(AVAILABILITY_NODE)) {
+            //zookeeper data corrupted
+            log.fatal("Received a zookeeper corruption message")    
+          } else if (path.startsWith(MEMBERSHIP_NODE) || path.startsWith(AVAILABILITY_NODE)) {
+            log.info("Received an event where a node was deleted:%s".format(path))
+          } else {
+            log.error("Node deleted unexpectedly %s".format(path))
+          }
+
+          case NodeCreated(path) => {
+            log.error("Received an unexpected create event:%s".format(path))
+          }
+
           case m => log.error("Received unknown message: %s".format(m))
         }
       }
@@ -461,6 +476,10 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
         case EventType.NodeChildrenChanged => zooKeeperManager ! NodeChildrenChanged(event.getPath)
 
         case EventType.NodeDataChanged => zooKeeperManager ! NodeDataChanged(event.getPath)
+
+        case EventType.NodeCreated => zooKeeperManager ! NodeCreated(event.getPath)
+
+        case EventType.NodeDeleted => zooKeeperManager ! NodeDeleted(event.getPath)
       }
     }
 
diff --git a/cluster/src/main/scala/com/linkedin/norbert/jmx/AverageTimeTracker.scala b/cluster/src/main/scala/com/linkedin/norbert/jmx/AverageTimeTracker.scala
index 602a9a5b..c4a99d88 100644
--- a/cluster/src/main/scala/com/linkedin/norbert/jmx/AverageTimeTracker.scala
+++ b/cluster/src/main/scala/com/linkedin/norbert/jmx/AverageTimeTracker.scala
@@ -71,6 +71,106 @@ class FinishedRequestTimeTracker(clock: Clock, interval: Long) {
   }
 }
 
+class TotalRequestProcessingTime[KeyT](clock:Clock, interval:Long) {
+  private val q = new java.util.concurrent.ConcurrentLinkedQueue[(Long, Long)]()
+  private val currentlyCleaning = new java.util.concurrent.atomic.AtomicBoolean
+
+  private def clean {
+    // Let only one thread clean at a time
+    if(currentlyCleaning.compareAndSet(false, true)) {
+      clean0
+      currentlyCleaning.set(false)
+    }
+  }
+
+  private def clean0 {
+    while(!q.isEmpty) {
+      val head = q.peek
+      if(head == null)
+        return
+
+      val (completion, processingTime) = head
+      if(clock.getCurrentTimeOffsetMicroseconds - completion > interval) {
+        q.remove(head)
+      } else {
+        return
+      }
+    }
+  }
+
+  def addTime(processingTime: Long) {
+    clean
+    q.offer( (clock.getCurrentTimeOffsetMicroseconds, processingTime) )
+  }
+
+  def getArray: Array[(Long, Long)] = {
+    clean
+    q.toArray(Array.empty[(Long, Long)])
+  }
+
+  def getTimings: Array[Long] = {
+    getArray.map(_._2).sorted
+  }
+
+  def total = {
+    getTimings.sum
+  }
+
+  def reset {
+    q.clear
+  }
+}
+
+class QueueTimeTracker[KeyT](clock: Clock, interval: Long) {
+    private val q = new java.util.concurrent.ConcurrentLinkedQueue[(Long, Long)]()
+    private val currentlyCleaning = new java.util.concurrent.atomic.AtomicBoolean
+
+    private def clean {
+      // Let only one thread clean at a time
+      if(currentlyCleaning.compareAndSet(false, true)) {
+        clean0
+        currentlyCleaning.set(false)
+      }
+    }
+
+    private def clean0 {
+      while(!q.isEmpty) {
+        val head = q.peek
+        if(head == null)
+          return
+
+        val (completion, processingTime) = head
+        if(clock.getCurrentTimeOffsetMicroseconds - completion > interval) {
+          q.remove(head)
+        } else {
+          return
+        }
+      }
+    }
+
+    def addTime(processingTime: Long) {
+      clean
+      q.offer( (clock.getCurrentTimeOffsetMicroseconds, processingTime) )
+    }
+
+    def getArray: Array[(Long, Long)] = {
+      clean
+      q.toArray(Array.empty[(Long, Long)])
+    }
+
+    def getTimings: Array[Long] = {
+      getArray.map(_._2).sorted
+    }
+
+    def total = {
+      getTimings.sum
+    }
+
+    def reset {
+      q.clear
+    }
+}
+
 // Threadsafe
 class PendingRequestTimeTracker[KeyT](clock: Clock) {
   private val numRequests = new AtomicInteger()
@@ -78,16 +178,24 @@ class PendingRequestTimeTracker[KeyT](clock: Clock) {
   private val map : java.util.concurrent.ConcurrentMap[KeyT, Long] =
     new java.util.concurrent.ConcurrentHashMap[KeyT, Long]
 
+  private val mapQueueTime : java.util.concurrent.ConcurrentMap[KeyT, Long] = 
+    new java.util.concurrent.ConcurrentHashMap[KeyT, Long]
+
   def getStartTime(key: KeyT) = Option(map.get(key))
 
-  def beginRequest(key: KeyT) {
+  //pre-condition for this method is the above method returns some 
+  def getQueueTime(key: KeyT) = map.get(key)
+
+  def beginRequest(key: KeyT, queueTime: Long) {
     numRequests.incrementAndGet
     val now = clock.getCurrentTimeOffsetMicroseconds
     map.put(key, now)
+    mapQueueTime.put(key, queueTime)
   }
 
   def endRequest(key: KeyT) {
     map.remove(key)
+    mapQueueTime.remove(key)
   }
 
   def getTimings = {
@@ -108,14 +216,21 @@ class PendingRequestTimeTracker[KeyT](clock: Clock) {
 class RequestTimeTracker[KeyT](clock: Clock, interval: Long) {
   val finishedRequestTimeTracker = new FinishedRequestTimeTracker(clock, interval)
   val pendingRequestTimeTracker = new PendingRequestTimeTracker[KeyT](clock)
+  val queueTimeTracker = new QueueTimeTracker[KeyT](clock, interval)//TODO
+  val totalRequestProcessingTimeTracker = new TotalRequestProcessingTime[KeyT](clock, interval)
 
-  def beginRequest(key: KeyT) {
-    pendingRequestTimeTracker.beginRequest(key)
+  def beginRequest(key: KeyT, queueTime: Long = 0) {
+    pendingRequestTimeTracker.beginRequest(key, queueTime)
   }
 
   def endRequest(key: KeyT) {
     pendingRequestTimeTracker.getStartTime(key).foreach { startTime =>
+      //over time we will retire this since this does not account for the amount of time the request
+      //was stuck in the queue
       finishedRequestTimeTracker.addTime(clock.getCurrentTimeOffsetMicroseconds - startTime)
+      val queueTime = pendingRequestTimeTracker.getQueueTime(key)
+      queueTimeTracker.addTime(queueTime)
+      totalRequestProcessingTimeTracker.addTime(queueTime + clock.getCurrentTimeOffsetMicroseconds - startTime)
     }
     pendingRequestTimeTracker.endRequest(key)
   }
@@ -123,5 +238,7 @@ class RequestTimeTracker[KeyT](clock: Clock, interval: Long) {
   def reset {
     finishedRequestTimeTracker.reset
     pendingRequestTimeTracker.reset
+    queueTimeTracker.reset
+    totalRequestProcessingTimeTracker.reset
   }
 }
diff --git a/cluster/src/test/scala/com/linkedin/norbert/jmx/AverageTimeTrackerSpec.scala b/cluster/src/test/scala/com/linkedin/norbert/jmx/AverageTimeTrackerSpec.scala
index c9894e7c..9f8cfb7a 100644
--- a/cluster/src/test/scala/com/linkedin/norbert/jmx/AverageTimeTrackerSpec.scala
+++ b/cluster/src/test/scala/com/linkedin/norbert/jmx/AverageTimeTrackerSpec.scala
@@ -40,10 +40,10 @@ class AverageTimeTrackerSpec extends Specification {
 
        (0 until 10).foreach { i =>
          MockClock.currentTime = 1000L * i
-         tracker.beginRequest(i)
+         tracker.beginRequest(i,0)
          (tracker.total / (i + 1)) must be_==(1000L * i / 2)
        }
     }
 
   }
-}
\ No newline at end of file
+}
diff --git a/defaultEnvironment.gradle b/defaultEnvironment.gradle
new file mode 100644
index 00000000..db550d29
--- /dev/null
+++ b/defaultEnvironment.gradle
@@ -0,0 +1,6 @@
+subprojects {
+  repositories {
+    mavenCentral()
+  }
+}
+
diff --git a/examples/src/main/resources/log4j.properties b/examples/src/main/resources/log4j.properties
index 974a6b56..3de23a37 100644
--- a/examples/src/main/resources/log4j.properties
+++ b/examples/src/main/resources/log4j.properties
@@ -3,11 +3,11 @@ log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
 log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
 log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} - %-5p [%t:%C{1}@%L] - %m%n
 
-#log4j.appender.R=org.apache.log4j.RollingFileAppender
-#log4j.appender.R.File=/tmp/norbert.log
-#log4j.appender.R.MaxFileSize=100KB
-#log4j.appender.R.MaxBackupIndex=1
-#log4j.appender.R.layout=org.apache.log4j.PatternLayout
-#log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n
+log4j.appender.R=org.apache.log4j.RollingFileAppender
+log4j.appender.R.File=/tmp/norbert.log
+log4j.appender.R.MaxFileSize=100KB
+log4j.appender.R.MaxBackupIndex=1
+log4j.appender.R.layout=org.apache.log4j.PatternLayout
+log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n
 
-#log4j.category.com.linkedin.norbert=debug
+log4j.category.com.linkedin.norbert=debug
diff --git a/gradle.properties b/gradle.properties
new file mode 100644
index 00000000..f25a9ab7
--- /dev/null
+++ b/gradle.properties
@@ -0,0 +1 @@
+version=0.6.36
diff --git a/java-cluster/build.gradle b/java-cluster/build.gradle
new file mode 100644
index 00000000..c8793a67
--- /dev/null
+++ b/java-cluster/build.gradle
@@ -0,0 +1,11 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile project(':cluster')
+  compile externalDependency.scalaLibrary 
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+
diff --git a/java-cluster/src/main/scala/com/linkedin/norbert/javacompat.scala b/java-cluster/src/main/scala/com/linkedin/norbert/javacompat/package.scala
similarity index 99%
rename from java-cluster/src/main/scala/com/linkedin/norbert/javacompat.scala
rename to java-cluster/src/main/scala/com/linkedin/norbert/javacompat/package.scala
index 656bbc0b..84d2ec06 100644
--- a/java-cluster/src/main/scala/com/linkedin/norbert/javacompat.scala
+++ b/java-cluster/src/main/scala/com/linkedin/norbert/javacompat/package.scala
@@ -28,16 +28,16 @@ package object javacompat {
   implicit def javaSetToImmutableSet[T](nodes: java.util.Set[T]): Set[T] = {
     collection.JavaConversions.asScalaSet(nodes).foldLeft(Set[T]()) { (set, n) => set + n }
   }
-  
+
   implicit def javaIntegerSetToScalaIntSet(set: java.util.Set[java.lang.Integer]): Set[Int] = {
     collection.JavaConversions.asScalaSet(set).foldLeft(collection.immutable.Set.empty[Int]) { _ + _.intValue }
   }
-  
+
   implicit def scalaIntSetToJavaIntegerSet(set: Set[Int]): java.util.Set[java.lang.Integer] = {
     val result = new java.util.HashSet[java.lang.Integer](set.size)
     set.foreach (result add _)
     result
-  }  
+  }
 
   implicit def scalaNodeToJavaNode(node: SNode): JNode = {
     if (node == null) null else JavaNode(node)
diff --git a/java-network/build.gradle b/java-network/build.gradle
new file mode 100644
index 00000000..4fe88518
--- /dev/null
+++ b/java-network/build.gradle
@@ -0,0 +1,12 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile project(':network')
+  compile project(':java-cluster')
+  compile externalDependency.scalaLibrary 
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+
diff --git a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.java b/java-network/src/main/java/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.java
deleted file mode 100644
index 4acf6e84..00000000
--- a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Copyright 2009-2010 LinkedIn, Inc
- *
- * Licensed 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 com.linkedin.norbert.javacompat.network;
-
-public class PartitionedNetworkClientFactory<PartitionedId> {
-
-  private final NetworkClientConfig _config;
-  private final PartitionedLoadBalancerFactory<PartitionedId> _partitionedLoadBalancerFactory;
-
-  public PartitionedNetworkClientFactory(String serviceName,
-                                           String zooKeeperConnectString,
-                                           Integer zooKeeperSessionTimeoutMillis,
-                                           Long closeChannelTimeMillis,
-                                           Double norbertOutlierMultiplier,
-                                           Double norbertOutlierConstant,
-                                           PartitionedLoadBalancerFactory<PartitionedId> partitionedLoadBalancerFactory)
-  {
-    _config = new NetworkClientConfig();
-    _config.setServiceName(serviceName);
-    _config.setZooKeeperConnectString(zooKeeperConnectString);
-    _config.setZooKeeperSessionTimeoutMillis(zooKeeperSessionTimeoutMillis);
-    _config.setCloseChannelTimeMillis(closeChannelTimeMillis);
-    _config.setOutlierMuliplier(norbertOutlierMultiplier);
-    _config.setOutlierConstant(norbertOutlierConstant);
-    _partitionedLoadBalancerFactory = partitionedLoadBalancerFactory;
-  }
-
-
-  public PartitionedNetworkClient<PartitionedId> createPartitionedNetworkClient()
-  {
-    return new NettyPartitionedNetworkClient<PartitionedId>(_config,
-                                                       _partitionedLoadBalancerFactory);
-  }
-}
diff --git a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancer.java b/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancer.java
deleted file mode 100644
index 1cda631c..00000000
--- a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancer.java
+++ /dev/null
@@ -1,114 +0,0 @@
-///*
-//* Copyright 2009-2010 LinkedIn, Inc
-//*
-//* Licensed 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 com.linkedin.norbert.javacompat.network;
-//
-//import java.util.*;
-//
-//import org.apache.log4j.Logger;
-//
-//import com.linkedin.norbert.javacompat.cluster.Node;
-//
-//public class RingHashLoadBalancer implements PartitionedLoadBalancer<Integer> {
-//  private static Logger logger = Logger.getLogger(RingHashLoadBalancer.class);
-//
-//  private final TreeMap<Long, int[]> nodeCircleMap;
-//  private final HashFunction<Integer> hashStrategy;
-//  private final Set<Endpoint> endpoints;
-//  private final Random random = new java.util.Random();
-//
-//  private static Map<Integer, Set<Endpoint>> invert(Set<Endpoint> endpoints) {
-//    HashMap<Integer, Set<Endpoint>> pMap = new HashMap<Integer, Set<Endpoint>>();
-//
-//    for(Endpoint endpoint : endpoints) {
-//      Node node = endpoint.getNode();
-//
-//      for(Integer partitionId : node.getPartitionIds()) {
-//        Set<Endpoint> partitionEndpoints = pMap.get(partitionId) == null ? new HashSet<Endpoint>() : pMap.get(partitionId);
-//        partitionEndpoints.add(endpoint);
-//        pMap.put(partitionId, partitionEndpoints);
-//      }
-//    }
-//    return pMap;
-//  }
-//
-//  private static <K, V> int getSize(Map<K, ? extends Collection<V>> map) {
-//    int size = 0;
-//    for(Collection<V> v : map.values()) size += v.size();
-//    return size;
-//  }
-//
-//  public static RingHashLoadBalancer build(HashFunction<Integer> hashStrategy, int numReplicas, Set<Endpoint> endpoints) {
-//    Map<Integer, Set<Endpoint>> pMap = invert(endpoints);
-//    TreeMap<Long, int[]> nodeCircleMap = new TreeMap<Long, int[]>();
-//
-//    int slots = numReplicas * getSize(pMap);
-//
-//    Collection<Integer> partitions = pMap.keySet();
-//    for(int r = 0; r < slots; r++) {
-//      int[] nodeGroup = new int[partitions.size()];
-//      int i = 0;
-//      for(Integer partition : partitions) {
-//        nodeGroup[i++] = r % pMap.get(partition).size();
-//      }
-//
-//      String distKey = r + Arrays.toString(nodeGroup);
-//      nodeCircleMap.put(hashStrategy.hash(distKey), nodeGroup);
-//    }
-//
-//    return new RingHashLoadBalancer(hashStrategy, endpoints, nodeCircleMap);
-//  }
-//
-//  private RingHashLoadBalancer(HashFunction<Integer> hashStrategy, Set<Endpoint> endpoints, TreeMap<Long, int[]> nodeCircleMap) {
-//    this.hashStrategy = hashStrategy;
-//    this.endpoints = endpoints;
-//    this.nodeCircleMap = nodeCircleMap;
-//  }
-//
-//  public Set<Integer> getPartitions() {
-//    Set<Integer> result = new HashSet<Integer>();
-//    for(Endpoint e : endpoints) {
-//      result.addAll(e.getNode().getPartitionIds());
-//    }
-//    return result;
-//  }
-//
-//  public int[] getNodeGroup(Integer id) {
-//    if (nodeCircleMap.isEmpty())
-//      return null;
-//
-//    Long hash = hashStrategy.hash(id.hashCode());
-//    hash = nodeCircleMap.ceilingKey(hash);
-//    hash = (hash == null) ? nodeCircleMap.firstKey() : hash;
-//
-//    int[] nodegroup = nodeCircleMap.get(hash);
-//    if (logger.isDebugEnabled()) {
-//      logger.debug(id + " is sent to node group " + Arrays.toString(nodegroup));
-//    }
-//    return nodegroup;
-//  }
-//
-//  @Override
-//  public Node nextNode(Integer id) {
-//    int[] nodeGroup = getNodeGroup(id);
-//    int idx = random.nextInt(nodeGroup.length);
-//    return nodeGroup[i]
-//  }
-//
-//  @Override
-//  public Map<Node, Set<Integer>> nodesForOneReplica() {
-//
-//  }
-//}
diff --git a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancerFactory.java b/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancerFactory.java
deleted file mode 100644
index 0667c094..00000000
--- a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RingHashLoadBalancerFactory.java
+++ /dev/null
@@ -1,41 +0,0 @@
-///*
-//* Copyright 2009-2010 LinkedIn, Inc
-//*
-//* Licensed 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 com.linkedin.norbert.javacompat.network;
-//
-//import java.util.Set;
-//
-//import com.linkedin.norbert.cluster.InvalidClusterException;
-//import org.apache.log4j.Logger;
-//
-//import com.linkedin.norbert.javacompat.cluster.Node;
-//
-//public class RingHashLoadBalancerFactory implements PartitionedLoadBalancerFactory<Integer>
-//{
-//  private static Logger logger = Logger.getLogger(RingHashLoadBalancerFactory.class);
-//  private final int _numberOfReplicas;
-//  private final HashFunction<String> _hashingStrategy;
-//
-//  public RingHashLoadBalancerFactory(HashFunction<String> hashing, int numRep)
-//  {
-//    _numberOfReplicas = numRep;
-//    _hashingStrategy = hashing;
-//  }
-//
-//  @Override
-//  public PartitionedLoadBalancer<Integer> newLoadBalancer(Set<Endpoint> endpoints) throws InvalidClusterException {
-//    return new RingHashLoadBalancer(_hashingStrategy, _numberOfReplicas, endpoints);
-//  }
-//}
diff --git a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.java b/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.java
deleted file mode 100644
index afffe895..00000000
--- a/java-network/src/main/java/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Copyright 2009-2010 LinkedIn, Inc
- *
- * Licensed 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 com.linkedin.norbert.javacompat.network;
-
-import com.linkedin.norbert.EndpointConversions;
-import com.linkedin.norbert.cluster.InvalidClusterException;
-import com.linkedin.norbert.javacompat.cluster.JavaNode;
-import com.linkedin.norbert.javacompat.cluster.Node;
-import scala.Option;
-import scala.Some;
-
-import java.util.Set;
-
-public class RoundRobinLoadBalancerFactory implements LoadBalancerFactory {
-    private final com.linkedin.norbert.network.client.loadbalancer.RoundRobinLoadBalancerFactory scalaLbf =
-            new com.linkedin.norbert.network.client.loadbalancer.RoundRobinLoadBalancerFactory();
-
-    @Override
-    public LoadBalancer newLoadBalancer(Set<Endpoint> endpoints) throws InvalidClusterException {
-      final com.linkedin.norbert.network.client.loadbalancer.LoadBalancer loadBalancer =
-        scalaLbf.newLoadBalancer(EndpointConversions.convertJavaEndpointSet(endpoints));
-
-      return new LoadBalancer() {
-        @Override
-        public Node nextNode() {
-          return nextNode(0L, 0L);
-        }
-
-        @Override
-	public Node nextNode(Long capability) {
-	  return nextNode(capability,0L);	
-	}
-
-        @Override
-        public Node nextNode(Long capability, Long persistentCapability){
-          Option<com.linkedin.norbert.cluster.Node> node = loadBalancer.nextNode(new Some<Long>(capability.longValue()), new Some<Long>(persistentCapability.longValue()));
-          if(node.isDefined())
-            return JavaNode.apply(node.get());
-          else
-            return null;
-        }
-
-      };
-    }
-}
diff --git a/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.scala b/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.scala
new file mode 100644
index 00000000..14bd1e96
--- /dev/null
+++ b/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/PartitionedNetworkClientFactory.scala
@@ -0,0 +1,25 @@
+package com.linkedin.norbert.javacompat.network
+
+/**
+ *
+ * @author Dmytro Ivchenko
+ */
+class PartitionedNetworkClientFactory[PartitionedId](serviceName: String, zooKeeperConnectString: String,
+                                                     zooKeeperSessionTimeoutMillis: Int,
+                                                     closeChannelTimeMillis: Long, norbertOutlierMultiplier: Double, norbertOutlierConstant: Double,
+                                                     partitionedLoadBalancerFactory: PartitionedLoadBalancerFactory[PartitionedId])
+{
+  val config = new NetworkClientConfig
+
+  config.setServiceName(serviceName);
+  config.setZooKeeperConnectString(zooKeeperConnectString);
+  config.setZooKeeperSessionTimeoutMillis(zooKeeperSessionTimeoutMillis);
+  config.setCloseChannelTimeMillis(closeChannelTimeMillis);
+  config.setOutlierMuliplier(norbertOutlierMultiplier);
+  config.setOutlierConstant(norbertOutlierConstant);
+
+  def createPartitionedNetworkClient(): PartitionedNetworkClient[PartitionedId] =
+  {
+    new NettyPartitionedNetworkClient[PartitionedId](config, partitionedLoadBalancerFactory);
+  }
+}
diff --git a/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.scala b/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.scala
new file mode 100644
index 00000000..ebc16600
--- /dev/null
+++ b/java-network/src/main/scala/com/linkedin/norbert/javacompat/network/RoundRobinLoadBalancerFactory.scala
@@ -0,0 +1,38 @@
+package com.linkedin.norbert.javacompat.network
+
+
+import java.util
+import com.linkedin.norbert.EndpointConversions
+import com.linkedin.norbert.javacompat.cluster.{JavaNode, Node}
+
+
+/**
+ *
+ * @author Dmytro Ivchenko
+ */
+class RoundRobinLoadBalancerFactory extends LoadBalancerFactory
+{
+  val scalaLbf = new com.linkedin.norbert.network.client.loadbalancer.RoundRobinLoadBalancerFactory
+
+  def newLoadBalancer(endpoints: util.Set[Endpoint]): LoadBalancer = {
+    val loadBalancer = scalaLbf.newLoadBalancer(EndpointConversions.convertJavaEndpointSet(endpoints))
+
+    new LoadBalancer {
+      def nextNode: Node = {
+        nextNode(0L, 0L)
+      }
+
+      def nextNode(capability: java.lang.Long): Node = {
+        nextNode(capability, 0L)
+      }
+
+      def nextNode(capability: java.lang.Long, persistentCapability: java.lang.Long): Node = {
+        val node = loadBalancer.nextNode(new Some(capability.longValue()), new Some(persistentCapability.longValue()))
+        if (node.isDefined)
+          JavaNode.apply(node.get)
+        else
+          null
+      }
+    }
+  }
+}
diff --git a/build/sbt-launch.jar b/lib/sbt-launch.jar
similarity index 100%
rename from build/sbt-launch.jar
rename to lib/sbt-launch.jar
diff --git a/network/build.gradle b/network/build.gradle
new file mode 100644
index 00000000..978738e9
--- /dev/null
+++ b/network/build.gradle
@@ -0,0 +1,19 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile project(':cluster')
+  compile externalDependency.scalaLibrary 
+  compile externalDependency.netty
+  compile externalDependency.slf4jApi
+  compile externalDependency.slf4jLog4j
+
+  testCompile externalDependency.specs
+  testCompile externalDependency.mockitoAll
+  testCompile externalDependency.cglib
+  testCompile externalDependency.objenesis
+
+  scalaTools externalDependency.scalaCompiler
+  scalaTools externalDependency.scalaLibrary
+}
+
diff --git a/network/src/main/scala/com/linkedin/norbert/network/NetworkDefaults.scala b/network/src/main/scala/com/linkedin/norbert/network/NetworkDefaults.scala
index 7ccd273c..fa30ec9a 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/NetworkDefaults.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/NetworkDefaults.scala
@@ -100,7 +100,7 @@ object NetworkDefaults {
   /**
    * The default window size/time (in milliseconds) for averaging processing statistics
    */
-  val REQUEST_STATISTICS_WINDOW = 10000L
+  val REQUEST_STATISTICS_WINDOW = 10000000L
 
   /**
    *   Detects nodes that may be offline if their request processing times are greater than this multiplier over the average
@@ -110,7 +110,7 @@ object NetworkDefaults {
   /**
    * Detects nodes that may be offline if their request processing times are also greater than this additional constant
    */
-  val OUTLIER_CONSTANT = 10.0
+  val OUTLIER_CONSTANT = 10000.0
 
   /**
    * Protocol Buffers ByteString.copyFrom(byte[]) and ByteString.toByteArray both make a defensive copy of the
@@ -119,4 +119,9 @@ object NetworkDefaults {
    * if your JMV is running a Security Manager.
    */
   val AVOID_BYTESTRING_COPY = true
+
+  /**
+   * The default timeout for the time between starting iteration and the time we get the last element from iterator.
+   */
+  val DEFAULT_ITERATOR_TIMEOUT:Long = 5000
 }
diff --git a/network/src/main/scala/com/linkedin/norbert/network/client/NetworkClient.scala b/network/src/main/scala/com/linkedin/norbert/network/client/NetworkClient.scala
index 1fc7c62a..2f0dfb29 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/client/NetworkClient.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/client/NetworkClient.scala
@@ -24,6 +24,10 @@ import cluster._
 import network.common._
 import netty.NettyNetworkClient
 
+object NetworkClientConfig {
+  var defaultIteratorTimeout = NetworkDefaults.DEFAULT_ITERATOR_TIMEOUT;
+}
+
 class NetworkClientConfig {
   var clusterClient: ClusterClient = _
   var clientName: String = _
@@ -38,6 +42,8 @@ class NetworkClientConfig {
   var staleRequestTimeoutMins = NetworkDefaults.STALE_REQUEST_TIMEOUT_MINS
   var staleRequestCleanupFrequenceMins = NetworkDefaults.STALE_REQUEST_CLEANUP_FREQUENCY_MINS
 
+  var retryStrategy:Option[RetryStrategy] = None
+  var duplicatesOk:Boolean = false
 
   /**
    * Represents how long a channel stays alive. There are some specifics:
@@ -135,7 +141,7 @@ trait NetworkClient extends BaseNetworkClient {
   
   def sendRequest[RequestMsg, ResponseMsg](request: RequestMsg, capability: Option[Long], persistentCapability: Option[Long])
   (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): Future[ResponseMsg] = {
-    val future = new FutureAdapter[ResponseMsg]
+    val future = new FutureAdapterListener[ResponseMsg]
     sendRequest(request, future, capability, persistentCapability)
     future
   }
diff --git a/network/src/main/scala/com/linkedin/norbert/network/common/BaseNetworkClient.scala b/network/src/main/scala/com/linkedin/norbert/network/common/BaseNetworkClient.scala
index 0ea8c537..eb2d6875 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/common/BaseNetworkClient.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/common/BaseNetworkClient.scala
@@ -114,7 +114,7 @@ trait BaseNetworkClient extends Logging {
    */
   def sendRequestToNode[RequestMsg, ResponseMsg](request: RequestMsg, node: Node)
    (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): Future[ResponseMsg] = {
-    val future = new FutureAdapter[ResponseMsg]
+    val future = new FutureAdapterListener[ResponseMsg]
     sendRequestToNode(request, node, future)
     future
   }
diff --git a/network/src/main/scala/com/linkedin/norbert/network/common/NetworkStatisticsActor.scala b/network/src/main/scala/com/linkedin/norbert/network/common/NetworkStatisticsActor.scala
index e959407e..7fa3d63e 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/common/NetworkStatisticsActor.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/common/NetworkStatisticsActor.scala
@@ -53,9 +53,11 @@ class CachedNetworkStatistics[GroupIdType, RequestIdType](private val stats: Net
   val timings = CacheMaintainer(clock, refreshInterval, () => stats.getTimings)
   val pendingTimings = CacheMaintainer(clock, refreshInterval, () => stats.getPendingTimings)
   val totalRequests = CacheMaintainer(clock, refreshInterval, () => stats.getTotalRequests )
+  val finishedQueueTimings = CacheMaintainer(clock, refreshInterval, () => stats.getQueueTimings)
+  val responseTimings = CacheMaintainer(clock, refreshInterval, () => stats.getResponseTimings)
 
-  def beginRequest(groupId: GroupIdType, requestId: RequestIdType) {
-    stats.beginRequest(groupId, requestId)
+  def beginRequest(groupId: GroupIdType, requestId: RequestIdType, queueTime: Long) {
+    stats.beginRequest(groupId, requestId, queueTime)
   }
 
   def endRequest(groupId: GroupIdType, requestId: RequestIdType) {
@@ -82,7 +84,9 @@ class CachedNetworkStatistics[GroupIdType, RequestIdType](private val stats: Net
           pending = pendingTimings.get.map(calculate(_, p)).getOrElse(Map.empty),
           totalRequests = () => totalRequests.get.getOrElse(Map.empty),
           rps = () => finishedArray.get.map(_.mapValues(rps(_))).getOrElse(Map.empty),
-          requestQueueSize = () => finishedArray.get.map(_.mapValues(_.length)).getOrElse(Map.empty))
+          requestQueueSize = () => finishedArray.get.map(_.mapValues(_.length)).getOrElse(Map.empty),
+          finishedQueueTime = finishedQueueTimings.get.map(calculate(_, p)).getOrElse(Map.empty),
+          finishedResponse = responseTimings.get.map(calculate(_, p)).getOrElse(Map.empty))
       })
     }.get
   }
@@ -105,7 +109,9 @@ case class JoinedStatistics[K](finished: Map[K, StatsEntry],
                                pending: Map[K, StatsEntry],
                                rps: () => Map[K, Int],
                                totalRequests: () => Map[K, Int],
-                               requestQueueSize: () => Map[K, Int])
+                               requestQueueSize: () => Map[K, Int],
+                               finishedQueueTime: Map[K, StatsEntry],
+                               finishedResponse: Map[K, StatsEntry])
 
 private case class NetworkStatisticsTracker[GroupIdType, RequestIdType](clock: Clock, timeWindow: Long) extends Logging {
   private var timeTrackers: java.util.concurrent.ConcurrentMap[GroupIdType, RequestTimeTracker[RequestIdType]] =
@@ -115,8 +121,8 @@ private case class NetworkStatisticsTracker[GroupIdType, RequestIdType](clock: C
     atomicCreateIfAbsent(timeTrackers, groupId) { k => new RequestTimeTracker(clock, timeWindow) }
   }
 
-  def beginRequest(groupId: GroupIdType, requestId: RequestIdType) {
-    getTracker(groupId).beginRequest(requestId)
+  def beginRequest(groupId: GroupIdType, requestId: RequestIdType, queueTime:Long = 0) {
+    getTracker(groupId).beginRequest(requestId, queueTime)
   }
 
   def endRequest(groupId: GroupIdType, requestId: RequestIdType) {
@@ -140,6 +146,15 @@ private case class NetworkStatisticsTracker[GroupIdType, RequestIdType](clock: C
   }
 
   def getTotalRequests = timeTrackers.toMap.mapValues( _.pendingRequestTimeTracker.getTotalNumRequests )
+
+  //this does not need to be sorted since we are not doing 90th and 99th percentile
+  def getQueueTimings = {
+    timeTrackers.toMap.mapValues(_.queueTimeTracker.getTimings)
+  }
+
+  def getResponseTimings = {
+    timeTrackers.toMap.mapValues(_.totalRequestProcessingTimeTracker.getTimings)
+  }
 }
 
 trait NetworkClientStatisticsMBean {
@@ -184,23 +199,24 @@ class NetworkClientStatisticsMBeanImpl(clientName: Option[String], serviceName:
 
   private def getPendingStats(p: Double) = stats.getStatistics(p).map(_.pending).getOrElse(Map.empty)
   private def getFinishedStats(p: Double) = stats.getStatistics(p).map(_.finished).getOrElse(Map.empty)
+  private def toMillis(statsMetric: Double):Double = statsMetric/1000
 
   def getNumPendingRequests = toJMap(getPendingStats(0.5).map(kv => (kv._1.id, kv._2.size)))
 
   def getMedianTimes =
-    toJMap(getFinishedStats(0.5).map(kv => (kv._1.id, kv._2.percentile)))
+    toJMap(getFinishedStats(0.5).map(kv => (kv._1.id, toMillis(kv._2.percentile))))
 
   def get75thTimes =
-    toJMap(getFinishedStats(0.75).map(kv => (kv._1.id, kv._2.percentile)))
+    toJMap(getFinishedStats(0.75).map(kv => (kv._1.id, toMillis(kv._2.percentile))))
 
   def get90thTimes =
-    toJMap(getFinishedStats(0.90).map(kv => (kv._1.id, kv._2.percentile)))
+    toJMap(getFinishedStats(0.90).map(kv => (kv._1.id, toMillis(kv._2.percentile))))
 
   def get95thTimes =
-    toJMap(getFinishedStats(0.95).map(kv => (kv._1.id, kv._2.percentile)))
+    toJMap(getFinishedStats(0.95).map(kv => (kv._1.id, toMillis(kv._2.percentile))))
 
   def get99thTimes =
-    toJMap(getFinishedStats(0.99).map(kv => (kv._1.id, kv._2.percentile)))
+    toJMap(getFinishedStats(0.99).map(kv => (kv._1.id, toMillis(kv._2.percentile))))
 
   def getHealthScoreTimings = {
     val s = stats.getStatistics(0.5)
@@ -209,7 +225,7 @@ class NetworkClientStatisticsMBeanImpl(clientName: Option[String], serviceName:
 
     toJMap(f.map { case (n, nodeN) =>
       val nodeP = p.get(n).getOrElse(StatsEntry(0.0, 0, 0))
-      (n.id, doCalculation(Map(0 -> nodeP),Map(0 -> nodeN)))
+      (n.id, toMillis(doCalculation(Map(0 -> nodeP),Map(0 -> nodeN))))
     })
   }
 
@@ -222,23 +238,23 @@ class NetworkClientStatisticsMBeanImpl(clientName: Option[String], serviceName:
     val total = s.values.map(_.total).sum
     val size = s.values.map(_.size).sum
 
-    safeDivide(total, size)(0.0)
+    toMillis(safeDivide(total, size)(0.0))
   }
 
   def getClusterPendingTime = {
     val s = getPendingStats(0.5)
-    s.values.map(_.total).sum
+    toMillis(s.values.map(_.total).sum)
   }
 
-  def getClusterMedianTime = averagePercentiles(getFinishedStats(0.5))
+  def getClusterMedianTime = toMillis(averagePercentiles(getFinishedStats(0.5)))
 
-  def getCluster75thTimes = averagePercentiles(getFinishedStats(0.75))
+  def getCluster75thTimes = toMillis(averagePercentiles(getFinishedStats(0.75)))
 
-  def getCluster90th = averagePercentiles(getFinishedStats(0.90))
+  def getCluster90th = toMillis(averagePercentiles(getFinishedStats(0.90)))
 
-  def getCluster95th = averagePercentiles(getFinishedStats(0.95))
+  def getCluster95th = toMillis(averagePercentiles(getFinishedStats(0.95)))
 
-  def getCluster99th = averagePercentiles(getFinishedStats(0.99))
+  def getCluster99th = toMillis(averagePercentiles(getFinishedStats(0.99)))
 
   import scala.collection.JavaConversions._
 
@@ -248,7 +264,7 @@ class NetworkClientStatisticsMBeanImpl(clientName: Option[String], serviceName:
 
   def getClusterTotalRequests = getTotalRequests.values.sum
 
-  def getClusterHealthScoreTiming = doCalculation(getPendingStats(0.5), getFinishedStats(0.5))
+  def getClusterHealthScoreTiming = toMillis(doCalculation(getPendingStats(0.5), getFinishedStats(0.5)))
 
   def getQueueSize = stats.getStatistics(0.5).map(_.requestQueueSize().values.sum) getOrElse(0)
 
diff --git a/network/src/main/scala/com/linkedin/norbert/network/common/NorbertFuture.scala b/network/src/main/scala/com/linkedin/norbert/network/common/NorbertFuture.scala
index 3fcc550d..248e71b6 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/common/NorbertFuture.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/common/NorbertFuture.scala
@@ -18,9 +18,14 @@ package network
 package common
 
 import java.util.concurrent._
-import atomic.AtomicInteger
+import atomic.{AtomicLong, AtomicInteger}
 import logging.Logging
 import annotation.tailrec
+import partitioned.PartitionedNetworkClient
+import java.util
+import com.sun.xml.internal.ws.resources.SoapMessages
+import com.linkedin.norbert.cluster.Node
+import com.linkedin.norbert.network.client.NetworkClientConfig
 
 class ResponseQueue[ResponseMsg] extends java.util.concurrent.LinkedBlockingQueue[Either[Throwable, ResponseMsg]] {
   def += (res: Either[Throwable, ResponseMsg]): ResponseQueue[ResponseMsg] = {
@@ -29,9 +34,69 @@ class ResponseQueue[ResponseMsg] extends java.util.concurrent.LinkedBlockingQueu
   }
 }
 
+trait ListenableFuture {
+  def addListener(listener:Runnable, executor:Executor)
+}
+
+/**
+ *
+ * If the underlying async request completes with success then we will get onCompleted handler invoked
+ * Else the onThrowable handler will get invoked. Timeouts need to be handled outside of this.
+ */
+abstract class PromiseListener[ResponseMsg] {
+  /**
+   * Depending on the timing this method   
+   * could execute in the calling thread
+   * of addListener or in the norbert client
+   * thread. Fork another thread if this
+   * is going to be a heavy call.  
+   */
+  def onCompleted(response: ResponseMsg):Unit = {
+  }
+  /**
+   * Depending on the timing this method 
+   * could execute in the calling thread 
+   * of addListener or in the norbert client       
+   * thread. Fork another thread if this
+   * is going to be a heavy call.
+   */
+  def onThrowable(t: Throwable):Unit = {
+  }
+}
+
+class FutureAdapterListener[ResponseMsg] extends FutureAdapter[ResponseMsg] {
+  @volatile var mListener: PromiseListener[ResponseMsg] = null
+  def addListener(listener: PromiseListener[ResponseMsg]) {
+    synchronized {
+      mListener = listener
+      if(isDone)  {
+        response match {
+          case Left(t) => listener.onThrowable(t)
+          case Right(response) => listener.onCompleted(response)
+          case _ => listener.onThrowable(new IllegalStateException("Response was neither throwable nor an exception"))
+        }
+      }
+    }
+  }
+
+  override def apply(callback: Either[Throwable, ResponseMsg]): Unit = {
+    super.apply(callback)
+    synchronized {
+      if(mListener != null) {
+        response match {
+          case Left(t) => mListener.onThrowable(t)
+          case Right(response) => mListener.onCompleted(response)
+          case _ => mListener.onThrowable(new IllegalStateException("Response was neither throwable nor an exception"))
+        }
+      }
+    }
+  }
+}
+
+
 class FutureAdapter[ResponseMsg] extends Future[ResponseMsg] with Function1[Either[Throwable, ResponseMsg], Unit] with ResponseHelper {
-  private val latch = new CountDownLatch(1)
-  @volatile private var response: Either[Throwable, ResponseMsg] = null
+  protected val latch = new CountDownLatch(1)
+  @volatile protected var response: Either[Throwable, ResponseMsg] = null
 
   override def apply(callback: Either[Throwable, ResponseMsg]): Unit = {
     response = callback
@@ -193,6 +258,242 @@ case class PartialIterator[ResponseMsg](inner: ExceptionIterator[ResponseMsg]) e
   }
 }
 
+/**
+ * This class encapsulates the different parameters which can be used to configure selective retry.
+ * It also provides a hook onTimeout which can be overridden for tracking/logging purposes.
+ * @param timeoutForRetry This value specifies the retry timeout in milliseconds (amount of time post retry to wait)
+ * @param thresholdNodeFailures This value specifies the threshold of number of failed nodes which will be tolerated
+ * @param nextRetryStrategy This value specifies if there is another retry which we want to specify should prev fail
+ * @param initialTimeout This value is used to determine the very first timeout before the first retry kicks in
+ */
+class RetryStrategy(val timeoutForRetry: Long, val thresholdNodeFailures: Int, val nextRetryStrategy: Option[RetryStrategy] = None, val initialTimeout:Long=5000) {//TODO make constant in networkClientConfig
+  /**
+   * This method is a callback which we register with the iterator and is invoked on timeout
+   * @param numNodeFailures total number of nodes which have failed thus far
+   * @return Setup a future retry in case this retry fails
+   */
+  def onTimeout(numNodeFailures: Int):Option[RetryStrategy] = {
+    if(numNodeFailures <= thresholdNodeFailures) {
+      return nextRetryStrategy
+    }
+    val x = NetworkClientConfig.defaultIteratorTimeout
+    throw new TimeoutException("The number of responses which timed out from nodes exceeded threshold:%d requests".format(numNodeFailures))
+  }
+}
+
+/**
+ * The selective retry iterator essentially retries sub-requests which are part of larger request on timeout.
+ * @param numRequests The number of different norbert servers this request has been sent to initially
+ * @param timeoutForRetry The threshold of time for this retry attempt for which we are willing to wait
+ * @param sendRequestFunctor This is an opaque function used to send the requests out during retry
+ * @param setRequests This is a mapping from partition to node to track failed nodes and partitions to be retried.
+ * @param queue This is response queue in which we store tuples (node, request, response)
+ * @param calculateNodesFromIds This is an opaque function which calculates node -> partition mappings with exclusion list
+ * @param requestBuilder This is a functor as well
+ * @param is Serializer
+ * @param os Serializer
+ * @param retryStrategy This is the strategy to apply when we run into timeout situation.
+ * @param duplicatesOk Whether or not we can have duplicates returned to higher application layer.
+ * @tparam PartitionedId This is a type representing the partition id
+ * @tparam RequestMsg This is a type representing the request message type
+ * @tparam ResponseMsg This is a type representing the response message type
+ */
+class SelectiveRetryIterator[PartitionedId, RequestMsg, ResponseMsg](
+                              numRequests: Int, var timeoutForRetry: Long = 5000L,
+                              sendRequestFunctor: (PartitionedRequest[PartitionedId, RequestMsg, ResponseMsg] => Unit),
+                              var setRequests: Map[PartitionedId, Node],
+                              queue: ResponseQueue[Tuple3[Node, Set[PartitionedId], ResponseMsg]],
+                              calculateNodesFromIds : ((Set[PartitionedId], Set[Node], Int) => scala.collection.immutable.Map[Node,Set[PartitionedId]]),
+                              requestBuilder: (Node, Set[PartitionedId]) => RequestMsg,
+                              is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg],
+                              var retryStrategy: Option[RetryStrategy], var duplicatesOk: Boolean = false)
+                                extends ResponseIterator[ResponseMsg] with ResponseHelper{
+  /**
+   * Set of nodes which have failed in sending a response back in time for this larger request
+   */
+  var failedNodes : Set[Node] = Set.empty[Node]
+  /**
+   * The time we started a pass which is either the first attempt or post a retry
+   */
+  var timeStartedPass : Long = System.currentTimeMillis()
+  /**
+   * The number of machines/nodes which have not responded so far
+   */
+  var distinctResponsesLeft: Int = numRequests
+
+  /**
+   * This function determines whether an entry received from the queue
+   * can be returned back to the invoking higher layer
+   * @param node norbert server machine
+   * @param pIds set of pids that we got a response for from the node
+   * @return true - this response should be returned to higher layer, false - if this response should not be returned
+   */
+  def isValidQueueEntry(node: Node, pIds: Set[PartitionedId]) : Boolean = {
+    if(!duplicatesOk) {
+      if(!failedNodes.contains(node))
+        return true
+    } else {
+      val iter: Iterator[PartitionedId] = pIds.iterator
+      while(iter.hasNext) {
+        val partitionedId = iter.next
+        if(setRequests.contains(partitionedId))
+          return true
+      }
+    }
+    return false
+  }
+
+  /**
+   * @return true if a response is available without blocking, false otherwise
+   */
+  def nextAvailable:Boolean = {
+    //go through the queue discarding the ones which are dups or in failed nodes
+    var entry: Tuple3[Node, Set[PartitionedId], ResponseMsg] = null
+    while(true) {
+      queue.poll(0, TimeUnit.MILLISECONDS) match {
+       case null => return false
+       case e: Throwable => throw e;
+       case f: Tuple3[Node, Set[PartitionedId], ResponseMsg] => entry = f
+       case _ => return false
+      } 
+      if(isValidQueueEntry(entry._1, entry._2))
+        return true
+    }
+    return false
+  }
+
+  /**
+   *
+   * @param timeout how long to wait before giving up, in terms of <code>unit</code>
+   * @param unit the <code>TimeUnit</code> that <code>timeout</code> should be interpreted in
+   *
+   * @return a response
+   */
+  def next(timeout: Long, unit: TimeUnit):ResponseMsg = {
+    val timeEnded = System.currentTimeMillis + timeout
+    while(true) {
+      queue.poll(timeEnded - System.currentTimeMillis, unit) match {
+        case null =>
+          throw new TimeoutException("Timed out waiting for response")
+
+        case f => {
+          var e:Tuple3[Node, Set[PartitionedId], ResponseMsg] = null
+          f match {
+		case g:Throwable => throw g;
+                case h:Tuple3[Node, Set[PartitionedId], ResponseMsg] => e = h
+                case _ => return null.asInstanceOf[ResponseMsg] 
+          }
+          if(isValidQueueEntry(e._1, e._2)) {
+            return e._3
+          }
+        }
+      }
+    }
+    return null.asInstanceOf[ResponseMsg]
+  }
+
+  /**
+   * This method will block upto a time threshold for getting the next response.
+   * If a retry strategy has been setup we will at that point invoke a retry and continue waiting for responses.
+   * If we call next after reaching the end behavior is undefined
+   * @return a response
+   */
+  def next(): ResponseMsg = {
+    var timeoutCutoff: Long = timeStartedPass + timeoutForRetry
+    while(true) {
+      queue.poll(timeoutCutoff - System.currentTimeMillis(), TimeUnit.MILLISECONDS) match {
+        case null => {
+          var ids:scala.collection.immutable.Set[PartitionedId] = null
+          //if we have a retry strategy in place set new values
+          retryStrategy match {
+            case Some(e:RetryStrategy) => {
+
+              //iterate through the set to figure out the failed nodes for this pass as well
+              ids = setRequests.keySet.toSet
+              val remainingRequestsIterator : Iterator[Tuple2[PartitionedId, Node]] = setRequests.iterator
+
+              while(remainingRequestsIterator.hasNext) {
+                val tuple : Tuple2[PartitionedId, Node] = remainingRequestsIterator.next
+                failedNodes += tuple._2
+              }
+
+              //check if we meet the requirements for retry to occur or not
+              retryStrategy = e.onTimeout(failedNodes.size)
+              timeoutForRetry = e.timeoutForRetry
+
+              //time started pass should be relative the start time
+              timeStartedPass = timeoutCutoff
+              timeoutCutoff = timeStartedPass + timeoutForRetry 
+            }
+            case None => {
+              if (!duplicatesOk)
+                throw new TimeoutException("Timedout waiting for final %d nodes".format(distinctResponsesLeft))
+              else
+                throw new TimeoutException("Timedout waiting for final %d partitions to return".format(setRequests.size))
+            }
+          }
+
+	  //If for a particular partition id only if 10/10 of the replicas are in trouble then quit
+          val nodes = calculateNodesFromIds(ids, failedNodes, 10)
+
+          if(duplicatesOk != true) {
+            //only the responses from these new requests count
+            log.debug("Adjust responseIterator to: %d".format(nodes.keySet.size))
+            distinctResponsesLeft=nodes.keySet.size
+	    //reset the outstanding requests map
+	    setRequests = Map.empty[PartitionedId, Node]
+          }
+
+          nodes.foreach {
+            case (node, idsForNode) => {
+              def callback(a:Either[Throwable, ResponseMsg]):Unit = {
+                a match {
+                  case Left(t) => queue += Left(t)
+                  case Right(r) => queue += Right(Tuple3(node, idsForNode, r))
+                }
+              }
+
+              val request1 = PartitionedRequest(requestBuilder(node, idsForNode), node, idsForNode, requestBuilder, is, os, Some((a: Either[Throwable, ResponseMsg]) => {callback(a)}), 0, Some(this))
+	            idsForNode.foreach {
+		            case id => setRequests = setRequests + (id -> node)
+	            }
+              sendRequestFunctor(request1)
+            }
+          }
+        }
+
+        case e : Either[Throwable,Tuple3[Node, Set[PartitionedId], ResponseMsg]] => {
+          //check if we received an exception or response
+          e match {
+            case Right(response) => {
+              if(isValidQueueEntry(response._1, response._2)) {
+                distinctResponsesLeft=distinctResponsesLeft - 1
+                response._2.foreach {
+		  partitionId => setRequests -= partitionId
+		}
+                return response._3
+              }
+            }
+            case Left(exception) => {
+              throw exception
+            }
+          }
+        }
+        case _ => null.asInstanceOf[ResponseMsg]
+      }
+    }
+    null.asInstanceOf[ResponseMsg]
+  }
+
+  def hasNext = {
+    if(!duplicatesOk)
+      distinctResponsesLeft != 0
+    else {
+      setRequests.isEmpty != true
+    }
+  }
+}
+
 private[common] trait ResponseHelper extends Logging {
   protected def translateResponse[T](response: Either[Throwable, T]) = {
     val r = if(response == null) Left(new NullPointerException("Null response found"))
@@ -200,4 +501,4 @@ private[common] trait ResponseHelper extends Logging {
 
     r.fold(ex => throw new ExecutionException(ex) , msg => msg)
  }
-}
\ No newline at end of file
+}
diff --git a/network/src/main/scala/com/linkedin/norbert/network/netty/ClientChannelHandler.scala b/network/src/main/scala/com/linkedin/norbert/network/netty/ClientChannelHandler.scala
index 9bf1761e..1e7edef2 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/netty/ClientChannelHandler.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/netty/ClientChannelHandler.scala
@@ -102,7 +102,7 @@ class ClientChannelHandler(clientName: Option[String],
 
     if(!request.callback.isEmpty) {
       requestMap.put(request.id, request)
-      stats.beginRequest(request.node, request.id)
+      stats.beginRequest(request.node, request.id, 0)
     }
 
     val message = NorbertProtos.NorbertMessage.newBuilder
@@ -122,7 +122,9 @@ class ClientChannelHandler(clientName: Option[String],
     val requestId = new UUID(message.getRequestIdMsb, message.getRequestIdLsb)
 
     requestMap.get(requestId) match {
-      case null => log.warn("Received a response message [%s] without a corresponding request".format(message))
+      case null => {
+        log.warn("Received a response message UUID: [%s] without a corresponding request from %s".format(requestId, ctx.getChannel().getRemoteAddress()))
+      }
       case request =>
         requestMap.remove(requestId)
 
diff --git a/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkClient.scala b/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkClient.scala
index 71b28b10..07efc6ce 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkClient.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkClient.scala
@@ -130,4 +130,6 @@ abstract class BaseNettyNetworkClient(clientConfig: NetworkClientConfig) extends
 class NettyNetworkClient(clientConfig: NetworkClientConfig, val loadBalancerFactory: LoadBalancerFactory) extends BaseNettyNetworkClient(clientConfig) with NetworkClient with LoadBalancerFactoryComponent
 
 class NettyPartitionedNetworkClient[PartitionedId](clientConfig: NetworkClientConfig, val loadBalancerFactory: PartitionedLoadBalancerFactory[PartitionedId]) extends BaseNettyNetworkClient(clientConfig)
-    with PartitionedNetworkClient[PartitionedId] with PartitionedLoadBalancerFactoryComponent[PartitionedId]
\ No newline at end of file
+    with PartitionedNetworkClient[PartitionedId] with PartitionedLoadBalancerFactoryComponent[PartitionedId] {
+  setConfig(clientConfig)
+}
diff --git a/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkServer.scala b/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkServer.scala
index 2eddee38..628e179c 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkServer.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/netty/NettyNetworkServer.scala
@@ -37,6 +37,7 @@ class NetworkServerConfig {
   var zooKeeperSessionTimeoutMillis = 30000
 
   var requestTimeoutMillis = NetworkDefaults.REQUEST_TIMEOUT_MILLIS
+  var responseGenerationTimeoutMillis = -1//turned off by default
 
   var requestThreadCorePoolSize = NetworkDefaults.REQUEST_THREAD_CORE_POOL_SIZE
   var requestThreadMaxPoolSize = NetworkDefaults.REQUEST_THREAD_MAX_POOL_SIZE
@@ -62,7 +63,8 @@ class NettyNetworkServer(serverConfig: NetworkServerConfig) extends NetworkServe
                                                       maxPoolSize = serverConfig.requestThreadMaxPoolSize,
                                                       keepAliveTime = serverConfig.requestThreadKeepAliveTimeSecs,
                                                       maxWaitingQueueSize = serverConfig.threadPoolQueueSize,
-                                                      requestStatisticsWindow = serverConfig.requestStatisticsWindow)
+                                                      requestStatisticsWindow = serverConfig.requestStatisticsWindow,
+                                                      responseGenerationTimeoutMillis = serverConfig.responseGenerationTimeoutMillis)
 
   val executor = Executors.newCachedThreadPool(new NamedPoolThreadFactory("norbert-server-pool-%s".format(clusterClient.serviceName)))
   val bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(executor, executor))
diff --git a/network/src/main/scala/com/linkedin/norbert/network/netty/ServerChannelHandler.scala b/network/src/main/scala/com/linkedin/norbert/network/netty/ServerChannelHandler.scala
index 46bab1db..13ffa0c5 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/netty/ServerChannelHandler.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/netty/ServerChannelHandler.scala
@@ -116,7 +116,7 @@ class ServerChannelHandler(clientName: Option[String],
     val messageName = norbertMessage.getMessageName
     val requestBytes = ProtoUtils.byteStringToByteArray(norbertMessage.getMessage, avoidByteStringCopy)
 
-    statsActor.beginRequest(0, context.requestId)
+    statsActor.beginRequest(0, context.requestId, 0)
 
     val (handler, is, os) = try {
       val handler: Any => Any = messageHandlerRegistry.handlerFor(messageName)
@@ -142,7 +142,7 @@ class ServerChannelHandler(clientName: Option[String],
     catch {
       case ex: HeavyLoadException =>
         Channels.write(ctx, Channels.future(channel), (context, ResponseHelper.errorResponse(context.requestId, ex, NorbertProtos.NorbertMessage.Status.HEAVYLOAD)))
-        statsActor.endRequest(0, context.requestId)
+        statsActor.endRequest(0, context.requestId) 
     }
   }
 
@@ -187,24 +187,53 @@ trait NetworkServerStatisticsMBean {
   def getMedianTime: Double
   def get90PercentileTime: Double
   def get99PercentileTime: Double
+
+
+  def getAverageResponseProcessingTime: Double
+  def getMedianResponseTime: Double
+  def get90PercentileResponseTime: Double
+  def get99PercentileResponseTime: Double
 }
 
 class NetworkServerStatisticsMBeanImpl(clientName: Option[String], serviceName: String, val stats: CachedNetworkStatistics[Int, UUID])
   extends MBean(classOf[NetworkServerStatisticsMBean], JMX.name(clientName, serviceName)) with NetworkServerStatisticsMBean {
 
-  def getMedianTime = stats.getStatistics(0.5).map(_.finished.values.map(_.percentile)).flatten.sum
+  def toMillis(statsMetric: Double):Double = {statsMetric/1000} 
+
+  def getMedianTime = toMillis((stats.getStatistics(0.5).map(_.finished.values.map(_.percentile)).flatten.sum))
 
-  def getRequestsPerSecond = stats.getStatistics(0.5).map(_.rps().values).flatten.sum
+  def getRequestsPerSecond = toMillis((stats.getStatistics(0.5).map(_.rps().values).flatten.sum)).asInstanceOf[Int]
 
   def getAverageRequestProcessingTime = stats.getStatistics(0.5).map { stats =>
     val total = stats.finished.values.map(_.total).sum
     val size = stats.finished.values.map(_.size).sum
 
+    toMillis(safeDivide(total.toDouble, size)(0.0))
+  } getOrElse(0.0)
+
+  def get90PercentileTime = toMillis((stats.getStatistics(0.90).map(_.finished.values.map(_.percentile)).flatten.sum))
+
+  def get99PercentileTime = toMillis((stats.getStatistics(0.99).map(_.finished.values.map(_.percentile)).flatten.sum)) 
+
+  //the following statistics are in microseconds not milliseconds
+  def getAverageResponseProcessingTime = stats.getStatistics(0.5).map { stats =>
+    val total = stats.finishedResponse.values.map(_.total).sum
+    val size = stats.finishedResponse.values.map(_.size).sum
+
     safeDivide(total.toDouble, size)(0.0)
   } getOrElse(0.0)
 
-  def get90PercentileTime = stats.getStatistics(0.90).map(_.finished.values.map(_.percentile)).flatten.sum
+  def getMedianResponseTime = stats.getStatistics(0.5).map(_.finishedResponse.values.map(_.percentile)).flatten.sum
+
+  def get90PercentileResponseTime = stats.getStatistics(0.90).map(_.finishedResponse.values.map(_.percentile)).flatten.sum
 
-  def get99PercentileTime = stats.getStatistics(0.99).map(_.finished.values.map(_.percentile)).flatten.sum 
+  def get99PercentileResponseTime = stats.getStatistics(0.99).map(_.finishedResponse.values.map(_.percentile)).flatten.sum
+
+  def getAverageQueueTime = stats.getStatistics(0.5).map { stats =>
+    val total = stats.finishedQueueTime.values.map(_.total).sum
+    val size = stats.finishedQueueTime.values.map(_.size).sum
+
+    safeDivide(total.toDouble, size)(0.0)
+  } getOrElse (0.0)
 }
 
diff --git a/network/src/main/scala/com/linkedin/norbert/network/partitioned/PartitionedNetworkClient.scala b/network/src/main/scala/com/linkedin/norbert/network/partitioned/PartitionedNetworkClient.scala
index 21630185..8dffea89 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/partitioned/PartitionedNetworkClient.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/partitioned/PartitionedNetworkClient.scala
@@ -25,6 +25,8 @@ import netty.NettyPartitionedNetworkClient
 import client.NetworkClientConfig
 import cluster.{Node, ClusterDisconnectedException, InvalidClusterException, ClusterClientComponent}
 import scala.util.Random
+import java.util
+
 
 object PartitionedNetworkClient {
   def apply[PartitionedId](config: NetworkClientConfig, loadBalancerFactory: PartitionedLoadBalancerFactory[PartitionedId]): PartitionedNetworkClient[PartitionedId] = {
@@ -42,15 +44,23 @@ object PartitionedNetworkClient {
     nc.start
     nc
   }
-
 }
 
 /**
  * The network client interface for interacting with nodes in a partitioned cluster.
  */
 trait PartitionedNetworkClient[PartitionedId] extends BaseNetworkClient {
+
   this: ClusterClientComponent with ClusterIoClientComponent  with PartitionedLoadBalancerFactoryComponent[PartitionedId] =>
 
+  var duplicatesOk:Boolean = false
+  var retryStrategy:Option[RetryStrategy] = None
+  def setConfig(config:NetworkClientConfig): Unit = {
+    duplicatesOk = config.duplicatesOk
+    if(retryStrategy != null)
+      retryStrategy = config.retryStrategy
+  }
+
   @volatile private var loadBalancer: Option[Either[InvalidClusterException, PartitionedLoadBalancer[PartitionedId]]] = None
 
   def sendRequest[RequestMsg, ResponseMsg](id: PartitionedId, request: RequestMsg, callback: Either[Throwable, ResponseMsg] => Unit)
@@ -225,8 +235,12 @@ trait PartitionedNetworkClient[PartitionedId] extends BaseNetworkClient {
 
   def sendRequest[RequestMsg, ResponseMsg](ids: Set[PartitionedId], requestBuilder: (Node, Set[PartitionedId]) => RequestMsg, capability: Option[Long])
   (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): ResponseIterator[ResponseMsg] =
-    sendRequest(ids, requestBuilder, capability, None)
- 
+    sendRequest(ids, requestBuilder, 0, capability, None)
+
+  def sendRequest[RequestMsg, ResponseMsg](ids: Set[PartitionedId], requestBuilder: (Node, Set[PartitionedId]) => RequestMsg, capability: Option[Long], dupOk : Boolean)
+  (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): ResponseIterator[ResponseMsg] =
+    sendRequest(ids, requestBuilder, 0, capability, None, dupOk)
+
   def sendRequest[RequestMsg, ResponseMsg](ids: Set[PartitionedId], requestBuilder: (Node, Set[PartitionedId]) => RequestMsg, capability: Option[Long], persistentCapability: Option[Long])
   (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): ResponseIterator[ResponseMsg] = doIfConnected {
     sendRequest(ids, requestBuilder, 0, capability, persistentCapability)
@@ -259,20 +273,64 @@ trait PartitionedNetworkClient[PartitionedId] extends BaseNetworkClient {
   (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): ResponseIterator[ResponseMsg] = 
    sendRequest(ids, requestBuilder, maxRetry, capability, None)
 
-  def sendRequest[RequestMsg, ResponseMsg](ids: Set[PartitionedId], requestBuilder: (Node, Set[PartitionedId]) => RequestMsg, maxRetry: Int, capability: Option[Long], persistentCapability: Option[Long])
+  def sendRequest[RequestMsg, ResponseMsg](ids: Set[PartitionedId], requestBuilder: (Node, Set[PartitionedId]) => RequestMsg, maxRetry: Int, capability: Option[Long], persistentCapability: Option[Long], dupOk: Boolean = duplicatesOk)
                                           (implicit is: InputSerializer[RequestMsg, ResponseMsg], os: OutputSerializer[RequestMsg, ResponseMsg]): ResponseIterator[ResponseMsg] = doIfConnected {
     if (ids == null || requestBuilder == null) throw new NullPointerException
     val nodes = calculateNodesFromIds(ids, capability, persistentCapability)
-    val queue = new ResponseQueue[ResponseMsg]
-    val resIter = new NorbertDynamicResponseIterator[ResponseMsg](nodes.size, queue)
-    nodes.foreach { case (node, idsForNode) =>
-      try {
-        doSendRequest(PartitionedRequest(requestBuilder(node, idsForNode), node, idsForNode, requestBuilder, is, os, if (maxRetry == 0) Some((a: Either[Throwable, ResponseMsg]) => {queue += a :Unit}) else Some(retryCallback[RequestMsg, ResponseMsg](queue.+=, maxRetry, capability, persistentCapability)_), 0, Some(resIter)))
-      } catch {
-        case ex: Exception => queue += Left(ex)
+    if (nodes.size <= 1 || retryStrategy == None) {
+      val queue = new ResponseQueue[ResponseMsg]
+      val resIter = new NorbertDynamicResponseIterator[ResponseMsg](nodes.size, queue)
+      nodes.foreach { case (node, idsForNode) =>
+        try {
+          doSendRequest(PartitionedRequest(requestBuilder(node, idsForNode), node, idsForNode, requestBuilder, is, os, if (maxRetry == 0) Some((a: Either[Throwable, ResponseMsg]) => {queue += a :Unit}) else Some(retryCallback[RequestMsg, ResponseMsg](queue.+=, maxRetry, capability, persistentCapability)_), 0, Some(resIter)))
+        } catch {
+          case ex: Exception => queue += Left(ex)
+        }
       }
+      return resIter
+    } else {
+      val nodes = calculateNodesFromIds(ids, None, None)
+      var setRequests: Map[PartitionedId, Node] = Map.empty[PartitionedId, Node]
+      nodes.foreach {
+        case (node, pids) => {
+          pids.foreach{
+	          case(pid) => setRequests += pid->node
+          }
+        }
+      }
+      val queue = new ResponseQueue[Tuple3[Node, Set[PartitionedId], ResponseMsg]]
+
+      /* wrapper so that iterator does not have to care about capability stuff */
+      def calculateNodesFromIdsSRetry(ids: Set[PartitionedId], excludedNodes: Set[Node], maxAttempts: Int)
+                                      :Map[Node, Set[PartitionedId]] = {
+        calculateNodesFromIds(ids, excludedNodes, maxAttempts, capability, persistentCapability).toMap
+      }
+
+      val resIter = new SelectiveRetryIterator[PartitionedId, RequestMsg, ResponseMsg](
+                    nodes.size, retryStrategy.get.initialTimeout, doSendRequest, setRequests,
+                    queue, calculateNodesFromIdsSRetry, requestBuilder, is, os, retryStrategy,
+                    dupOk)
+
+      nodes.foreach {
+        case (node, idsForNode) => {
+          def callback(a:Either[Throwable, ResponseMsg]):Unit = {
+            a match {
+              case Left(t) => queue += Left(t)
+              case Right(r) => queue += Right(Tuple3(node, idsForNode, r))
+            }
+          }
+          try {
+            doSendRequest(PartitionedRequest(
+                          requestBuilder(node, idsForNode), node, idsForNode, requestBuilder, is, os,
+                          Some((a: Either[Throwable, ResponseMsg]) => {callback(a)}), 0, Some(resIter))
+                          )
+          } catch {
+            case ex: Exception => queue += Left(ex)
+          }
+        }
+      }
+      resIter
     }
-    resIter
   }
 
   /**
diff --git a/network/src/main/scala/com/linkedin/norbert/network/server/MessageExecutorComponent.scala b/network/src/main/scala/com/linkedin/norbert/network/server/MessageExecutorComponent.scala
index fd490fe7..467ec348 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/server/MessageExecutorComponent.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/server/MessageExecutorComponent.scala
@@ -17,6 +17,7 @@ package com.linkedin.norbert
 package network
 package server
 
+
 import logging.Logging
 import jmx.JMX.MBean
 import jmx.{FinishedRequestTimeTracker, JMX}
@@ -56,7 +57,8 @@ class ThreadPoolMessageExecutor(clientName: Option[String],
                                 maxPoolSize: Int,
                                 keepAliveTime: Int,
                                 maxWaitingQueueSize: Int,
-                                requestStatisticsWindow: Long) extends MessageExecutor with Logging {
+                                requestStatisticsWindow: Long,
+                                responseGenerationTimeoutMillis: Long) extends MessageExecutor with Logging {
   def this(clientName: Option[String],
            serviceName: String,
            messageHandlerRegistry: MessageHandlerRegistry,
@@ -65,14 +67,15 @@ class ThreadPoolMessageExecutor(clientName: Option[String],
            maxPoolSize: Int,
            keepAliveTime: Int,
            maxWaitingQueueSize: Int,
-           requestStatisticsWindow: Long) =
-    this(clientName, serviceName, messageHandlerRegistry, new MutableList[Filter], requestTimeout, corePoolSize, maxPoolSize, keepAliveTime, maxWaitingQueueSize, requestStatisticsWindow)
+           requestStatisticsWindow: Long,
+           responseGenerationTimeoutMillis: Long) =
+    this(clientName, serviceName, messageHandlerRegistry, new MutableList[Filter], requestTimeout, corePoolSize, maxPoolSize, keepAliveTime, maxWaitingQueueSize, requestStatisticsWindow, responseGenerationTimeoutMillis)
 
   private val statsActor = CachedNetworkStatistics[Int, Int](SystemClock, requestStatisticsWindow, 200L)
   private val totalNumRejected = new AtomicInteger
 
   val requestQueue = new ArrayBlockingQueue[Runnable](maxWaitingQueueSize)
-  val statsJmx = JMX.register(new RequestProcessorMBeanImpl(clientName, serviceName, statsActor, requestQueue))
+  val statsJmx = JMX.register(new RequestProcessorMBeanImpl(clientName, serviceName, statsActor, requestQueue, threadPool))
 
   private val threadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, requestQueue,
     new NamedPoolThreadFactory("norbert-message-executor")) {
@@ -80,7 +83,7 @@ class ThreadPoolMessageExecutor(clientName: Option[String],
     override def beforeExecute(t: Thread, r: Runnable) = {
       val rr = r.asInstanceOf[RequestRunner[_, _]]
 
-      statsActor.beginRequest(0, rr.id)
+      statsActor.beginRequest(0, rr.id, (System.currentTimeMillis() - rr.queuedAt) * 1000)
     }
 
     override def afterExecute(r: Runnable, t: Throwable) = {
@@ -135,6 +138,13 @@ class ThreadPoolMessageExecutor(clientName: Option[String],
           val handler = messageHandlerRegistry.handlerFor(request)
           try {
             val response = handler(request)
+            val timeResponse = System.currentTimeMillis - queuedAt
+            if(responseGenerationTimeoutMillis>0 && timeResponse > responseGenerationTimeoutMillis) {
+              totalNumRejected.incrementAndGet
+              log.warn("Request timed out by the time we generated response, ignoring! Currently = " + now + ". " +
+                "Queued at = " + queuedAt + ". Timeout = " + requestTimeout)
+              throw new Exception("Response took too long:%d".format(timeResponse))
+            }
             response match {
               case _:Unit => None
               case null => None
@@ -173,13 +183,17 @@ class ThreadPoolMessageExecutor(clientName: Option[String],
     def getMedianTime: Double
   }
 
-  class RequestProcessorMBeanImpl(clientName: Option[String], serviceName: String, val stats: CachedNetworkStatistics[Int, Int], queue: ArrayBlockingQueue[Runnable])
+  class RequestProcessorMBeanImpl(clientName: Option[String], serviceName: String, val stats: CachedNetworkStatistics[Int, Int], queue: ArrayBlockingQueue[Runnable], threadPool: ThreadPoolExecutor)
     extends MBean(classOf[RequestProcessorMBean], JMX.name(clientName, serviceName)) with RequestProcessorMBean {
     def getQueueSize = queue.size
 
     def getTotalNumRejected = totalNumRejected.get.abs
 
     def getMedianTime = stats.getStatistics(0.5).map(_.finished.values.map(_.percentile)).flatten.sum
+
+    def getCurrentPoolSize = threadPool.getPoolSize
+
+    def getActivePoolSize = threadPool.getActiveCount
   }
 }
 
diff --git a/network/src/main/scala/com/linkedin/norbert/network/server/NetworkServer.scala b/network/src/main/scala/com/linkedin/norbert/network/server/NetworkServer.scala
index 80b68c9b..718319a3 100644
--- a/network/src/main/scala/com/linkedin/norbert/network/server/NetworkServer.scala
+++ b/network/src/main/scala/com/linkedin/norbert/network/server/NetworkServer.scala
@@ -98,6 +98,8 @@ trait NetworkServer extends Logging {
           if (markAvailableWhenConnected) {
             log.debug("Marking node with id %d available".format(nodeId))
             try {
+              //clean up the state if any from a previously killed incarnation 
+              clusterClient.markNodeUnavailable(nodeId) 
               clusterClient.markNodeAvailable(nodeId, initialCapability)
             } catch {
               case ex: ClusterException => log.error(ex, "Unable to mark node available")
diff --git a/network/src/test/scala/com/linkedin/norbert/network/common/NorbertFutureSpec.scala b/network/src/test/scala/com/linkedin/norbert/network/common/NorbertFutureSpec.scala
index af137844..6f6ba348 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/common/NorbertFutureSpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/common/NorbertFutureSpec.scala
@@ -19,11 +19,29 @@ package common
 
 import org.specs.Specification
 import org.specs.mock.Mockito
-import java.util.concurrent.{TimeoutException, ExecutionException, TimeUnit}
+import java.util.concurrent.{TimeoutException, ExecutionException, TimeUnit, Executor, LinkedBlockingQueue, Executors}
 import scala.Right
 
+class CurrentThreadExecutor extends Executor {
+    def execute(r:Runnable) = {
+        r.run();
+    }
+}
+
 class NorbertFutureSpec extends Specification with Mockito with SampleMessage {
-  val future = new FutureAdapter[Ping]
+  case class ResponseExceptionWrapper(exception:ExecutionException, ping: Ping, isException:Boolean)
+  class Task(queue: LinkedBlockingQueue[ResponseExceptionWrapper]) extends PromiseListener[Ping] {
+    override def onCompleted(response: Ping):Unit = {
+      queue.offer(ResponseExceptionWrapper(null, response, false))
+    }
+    override def onThrowable(t: Throwable):Unit = {
+      queue.offer(ResponseExceptionWrapper(new ExecutionException(t), null, true))
+    }
+  }
+  //val future = new FutureAdapter[Ping]
+  val future = new FutureAdapterListener[Ping]
+  val queue = new LinkedBlockingQueue[ResponseExceptionWrapper]()
+  future.addListener(new Task(queue))
 
   "NorbertFuture" should {
     "not be done when created" in {
@@ -33,6 +51,20 @@ class NorbertFutureSpec extends Specification with Mockito with SampleMessage {
     "be done when value is set" in {
       future.apply(Right(new Ping))
       future.isDone must beTrue
+      queue.size must be(1)
+    }
+
+    "be done when value is set" in {
+      val future = new FutureAdapterListener[Ping]
+      val queue = new LinkedBlockingQueue[ResponseExceptionWrapper]()
+
+      val msg = new Ping
+      future.apply(Right(msg))
+      future.isDone must beTrue
+      queue.size must be(0)
+      future.addListener(new Task(queue))
+      queue.size must be(1)
+      queue.poll mustEqual ResponseExceptionWrapper(null, msg, false)
     }
 
     "return the value that is set" in {
@@ -51,6 +83,9 @@ class NorbertFutureSpec extends Specification with Mockito with SampleMessage {
       future.apply(Left(ex))
       future.get must throwA[ExecutionException]
       future.get(1, TimeUnit.MILLISECONDS) must throwA[ExecutionException]
+      queue.size must be(1)
+      val response: ResponseExceptionWrapper = queue.poll
+      response.isException mustEqual true
     }
   }
 }
diff --git a/network/src/test/scala/com/linkedin/norbert/network/common/NorbertResponseIteratorSpec.scala b/network/src/test/scala/com/linkedin/norbert/network/common/NorbertResponseIteratorSpec.scala
index e88b0be4..dbecc0ec 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/common/NorbertResponseIteratorSpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/common/NorbertResponseIteratorSpec.scala
@@ -5,11 +5,26 @@ package common
 import org.specs.Specification
 import org.specs.mock.Mockito
 import java.util.concurrent.{ExecutionException, TimeoutException, TimeUnit}
+import java.util
+import com.linkedin.norbert.cluster.Node
+import com.linkedin.norbert.network._
 
 class NorbertResponseIteratorSpec extends Specification with Mockito with SampleMessage {
   val responseQueue = new ResponseQueue[Ping]
   val it = new NorbertResponseIterator[Ping](2, responseQueue)
 
+  class DummyInputSerializer[X, Y] extends InputSerializer[X, Y]  {
+    def requestName: String = {"dummy"}
+    def requestFromBytes(bytes: Array[Byte]): X = {return null.asInstanceOf[X]}
+    def responseFromBytes(bytes: Array[Byte]): Y = {return null.asInstanceOf[Y]}
+  }
+
+  class DummyOutputSerializer[X, Y] extends OutputSerializer[X, Y]  {
+    def responseName: String = {"dummy"}
+    def requestToBytes(request: X): Array[Byte] = {return Array.empty[Byte]}
+    def responseToBytes(response: Y): Array[Byte] = {return Array.empty[Byte]}
+  }
+
   "NorbertResponseIterator" should {
 //    "return true for next until all responses have been consumed" in {
 //      it.hasNext must beTrue
@@ -79,5 +94,233 @@ class NorbertResponseIteratorSpec extends Specification with Mockito with Sample
 
       partialIterator.hasNext mustBe false
     }
+
+    //duplicates not ok
+    "if not timeout occurs behavior is unchanged in case of no timeout" in {
+      def calculateNodeFunctor(setPIds: Set[Int],setNodes: Set[Node],requestMsg: Int) = Map.empty[Node, Set[Int]]
+      def requestBuilderFunctor(node: Node, setPIds: Set[Int]):Int = 1
+      val queue = new ResponseQueue[Tuple3[Node, Set[Int], Int]]
+      queue += Right(Tuple3(Node(10, "endpoint", true), Set.empty[Int] + 100, 1000))
+      def callback(pRequest: PartitionedRequest[Int,Int,Int]) = queue += Right(Tuple3(null.asInstanceOf[Node], Set.empty[Int] + 1, 2000))
+      val iterator =  new SelectiveRetryIterator[Int, Int, Int](1, 50L,
+                                          callback, Map.empty[Int, Node],
+                                          queue,
+					                                calculateNodeFunctor,
+					                                requestBuilderFunctor,
+                                          new DummyInputSerializer[Int, Int],
+                                          new DummyOutputSerializer[Int, Int],
+                                          None)
+      //can we test this independently of partitioned network client
+      //insert stuff into the queue TODO
+      iterator.hasNext mustBe true
+      //queue.poll(0, TimeUnit.MILLISECONDS) mustEqual 1000
+      iterator.next mustEqual 1000
+
+      iterator.hasNext mustBe false
+    }
+
+    "if timeout occurs behavior is unchanged as compared to before" in {
+      def calculateNodeFunctor(setPIds: Set[Int],setNodes: Set[Node],requestMsg: Int) = Map.empty[Node, Set[Int]]
+      def requestBuilderFunctor(node: Node, setPIds: Set[Int]):Int = 1
+      val queue = new ResponseQueue[Tuple3[Node, Set[Int], Int]]
+      queue += Right(Tuple3(Node(10, "endpoint", true), Set.empty[Int] + 100, 1000))
+      queue += Right(Tuple3(Node(10, "endpoint", true), Set.empty[Int] + 200, 2000))
+      def callback(pRequest: PartitionedRequest[Int,Int,Int]) = queue += Right(Tuple3(null.asInstanceOf[Node], Set.empty[Int] + 100, 1000))
+      val iterator =  new SelectiveRetryIterator[Int, Int, Int](4, 50L,
+        callback, Map.empty[Int, Node],
+        queue,
+        calculateNodeFunctor,
+        requestBuilderFunctor,
+        new DummyInputSerializer[Int, Int],
+        new DummyOutputSerializer[Int, Int],
+        None)
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 1000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 2000
+
+      val insertQueueTimely = new Thread(new Runnable {
+        def run() {Thread.sleep(15); queue += Right(Tuple3(Node(10, "endpoint", true), Set.empty[Int] + 300, 3000)) }
+      })
+      insertQueueTimely.start()
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 3000
+
+      val insertQueueLate = new Thread(new Runnable {
+        def run() {Thread.sleep(50); queue += Right(Tuple3(Node(10, "endpoint", true), Set.empty[Int] + 400, 4000)) }
+      })
+      insertQueueLate.start()
+
+      var timeoutExceptionFlag = false
+      try {
+      iterator.hasNext mustBe true
+      iterator.next
+      } catch {
+        case e: TimeoutException => timeoutExceptionFlag= true
+      }
+      timeoutExceptionFlag mustBe true
+    }
+
+    "if timeout occurs we try to find the next node multiple times" in {
+      var invocationCount = 1
+      def calculateNodeFunctor(setPIds: Set[Int],setNodes: Set[Node],requestMsg: Int):Map[Node, Set[Int]] = {
+        if(invocationCount == 1) {
+          val failedNodes = Set.empty[Node] + Node(1, "node1", true) + Node(2, "node2", true)
+          setNodes.size mustEqual 2
+          setNodes mustEqual failedNodes
+          val partitionsRetry = Set.empty[Int] + 1 + 2 + 3 + 4 + 5 + 6
+          setPIds mustEqual partitionsRetry
+          return Map.empty[Node, Set[Int]] + (Node(4,"node4", true) -> (Set.empty[Int] + 1 + 2)) + (Node(5,"node5",true) -> (Set.empty[Int] + 3 + 4)) + (Node(6,"node6",true) -> (Set.empty[Int] + 5 + 6))
+        } else if(invocationCount == 2){
+          val failedNodes = Set.empty[Node] + Node(6, "node6", true)
+          setNodes.size mustEqual 1
+          setNodes mustEqual failedNodes
+          val partitionsRetry = Set.empty[Int] + 5 + 6
+          setPIds mustEqual partitionsRetry
+          return Map.empty[Node, Set[Int]] + (Node(7,"node7",true) -> (Set.empty[Int] + 5 + 6))
+        }
+        return Map.empty[Node, Set[Int]]
+      }
+
+      def requestBuilderFunctor(node: Node, setPIds: Set[Int]):Int = 1
+      val queue = new ResponseQueue[Tuple3[Node, Set[Int], Int]]
+      queue += Right(Tuple3(Node(3, "node3", true), Set.empty[Int] + 7 + 8 + 9, 1000))
+      def callback(pRequest: PartitionedRequest[Int,Int,Int]) = {
+        if(invocationCount == 1) {
+          val insertQueueLate = new Thread(new Runnable {
+            def run() {Thread.sleep(5000); queue += Right(Tuple3(Node(6, "endpoint", true), Set.empty[Int] + 5 + 6, 6000)) }
+          })
+          insertQueueLate.start()
+          val insertQueueLate1 = new Thread(new Runnable {
+            def run() {Thread.sleep(10); queue += Right(Tuple3(Node(4, "endpoint", true), Set.empty[Int] + 1 + 2, 4000)) }
+          })
+          insertQueueLate1.start()
+          val insertQueueLate2 = new Thread(new Runnable {
+            def run() {Thread.sleep(15); queue += Right(Tuple3(Node(5, "endpoint", true), Set.empty[Int] + 3 + 4, 5000)) }
+          })
+          insertQueueLate2.start()
+          invocationCount += 1
+        } else if(invocationCount == 2) {
+          val insertQueueLate = new Thread(new Runnable {
+            def run() {Thread.sleep(20); queue += Right(Tuple3(Node(7, "endpoint", true), Set.empty[Int] + 5 + 6, 9000)) }
+          })
+          insertQueueLate.start()
+        }
+      }
+      //requests should contain some partitions to node mapping
+      //if requests times out the selective retry case then we need to make
+      //sure that
+      val node3 = Node(3, "node3", true)
+      val node2 = Node(2, "node2", true)
+      val node1 = Node(1, "node1", true)
+      val iterator =  new SelectiveRetryIterator[Int, Int, Int](3, 10L,
+        callback, Map.empty[Int, Node] + (7 -> node3) + (8 -> node3) + (9 -> node3)
+          + (1 -> node1) + (2 -> node1) + (3 -> node1)
+          + (4 -> node2) + (5 -> node2) + (6 -> node2),
+        queue,
+        calculateNodeFunctor,
+        requestBuilderFunctor,
+        new DummyInputSerializer[Int, Int],
+        new DummyOutputSerializer[Int, Int],
+        Some(new RetryStrategy(100L,2,Some(new RetryStrategy(100L,3)))))
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 1000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 4000
+      invocationCount mustBe 2
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 5000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 9000
+      invocationCount mustBe 2
+    }
+
+    "in case of duplicates we should return duplicate values and stop as soon as we have complete partitions" in {
+      def calculateNodeFunctor(setPIds: Set[Int],setNodes: Set[Node],requestMsg: Int) = Map.empty[Node, Set[Int]]
+      def requestBuilderFunctor(node: Node, setPIds: Set[Int]):Int = 1
+      val queue = new ResponseQueue[Tuple3[Node, Set[Int], Int]]
+      queue += Right(Tuple3(Node(1, "endpoint", true), Set.empty[Int] + 1 + 2, 1000))
+      queue += Right(Tuple3(Node(2, "endpoint", true), Set.empty[Int] + 1 + 2, 2000))
+      queue += Right(Tuple3(Node(3, "endpoint", true), Set.empty[Int] + 1, 3000))
+      queue += Right(Tuple3(Node(4, "endpoint", true), Set.empty[Int] + 1 + 3, 4000))
+      queue += Right(Tuple3(Node(5, "endpoint", true), Set.empty[Int] + 4, 5000))
+      def callback(pRequest: PartitionedRequest[Int,Int,Int]) = queue += Right(Tuple3(null.asInstanceOf[Node], Set.empty[Int] + 100, 1000))
+      val node6 = Node(6, "node1", true)
+      val iterator =  new SelectiveRetryIterator[Int, Int, Int](4, 50L,
+        callback, Map.empty[Int, Node] + (1->node6) + (2->node6) + (3->node6) + (4->node6),
+        queue,
+        calculateNodeFunctor,
+        requestBuilderFunctor,
+        new DummyInputSerializer[Int, Int],
+        new DummyOutputSerializer[Int, Int],
+        None, true)
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 1000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 4000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 5000
+    }
+
+    "in case of duplicate outstanding requests the first one to complete wins" in {
+      def calculateNodeFunctor(setPIds: Set[Int],setNodes: Set[Node],requestMsg: Int):Map[Node, Set[Int]] = {
+        val failedNodes = Set.empty[Node] + Node(1, "node1", true) + Node(2, "node2", true)
+        setNodes.size mustEqual 2
+        setNodes mustEqual failedNodes
+        val partitionsRetry = Set.empty[Int] + 1 + 2
+        setPIds mustEqual partitionsRetry
+        return Map.empty[Node, Set[Int]] + (Node(3,"node3",true) -> (Set.empty[Int] + 1)) + (Node(4,"node4",true) -> (Set.empty[Int] + 2))
+      }
+
+      def requestBuilderFunctor(node: Node, setPIds: Set[Int]):Int = 1
+      val queue = new ResponseQueue[Tuple3[Node, Set[Int], Int]]
+      def callback(pRequest: PartitionedRequest[Int,Int,Int]) = {
+        val insertQueueLate = new Thread(new Runnable {
+          def run() {Thread.sleep(10L); queue += Right(Tuple3(Node(3, "endpoint", true), Set.empty[Int] + 1, 3000)) }
+        })
+        insertQueueLate.start()
+        val insertQueueLate1 = new Thread(new Runnable {
+          def run() {Thread.sleep(100); queue += Right(Tuple3(Node(4, "endpoint", true), Set.empty[Int] + 2, 4000)) }
+        })
+        insertQueueLate1.start()
+      }
+      val node1 = Node(1, "node1", true)
+      val node2 = Node(2, "node2", true)
+      val iterator =  new SelectiveRetryIterator[Int, Int, Int](2, 20L,
+        callback, Map.empty[Int, Node] + (1->node1) + (2->node2),
+        queue,
+        calculateNodeFunctor,
+        requestBuilderFunctor,
+        new DummyInputSerializer[Int, Int],
+        new DummyOutputSerializer[Int, Int],
+        Some(new RetryStrategy(200L, 2)), true)
+
+      val insertQueueLate1 = new Thread(new Runnable {
+        def run() {Thread.sleep(100); queue += Right(Tuple3(Node(1, "endpoint", true), Set.empty[Int] + 1, 1000)) }
+      })
+      insertQueueLate1.start()
+      val insertQueueLate2 = new Thread(new Runnable {
+        def run() {Thread.sleep(90); queue += Right(Tuple3(Node(2, "endpoint", true), Set.empty[Int] + 2, 2000)) }
+      })
+      insertQueueLate2.start()
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 3000
+
+      iterator.hasNext mustBe true
+      iterator.next mustEqual 2000
+
+      iterator.hasNext mustBe false
+    }
   }
-}
\ No newline at end of file
+}
diff --git a/network/src/test/scala/com/linkedin/norbert/network/netty/ChannelPoolSpec.scala b/network/src/test/scala/com/linkedin/norbert/network/netty/ChannelPoolSpec.scala
index 2998ddcc..e70e664d 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/netty/ChannelPoolSpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/netty/ChannelPoolSpec.scala
@@ -321,5 +321,12 @@ class ChannelPoolSpec extends Specification with Mockito {
     def setSuccess = false
 
     def setProgress(p1: Long, p2: Long, p3: Long) = false
+    
+    def syncUninterruptibly = null
+    
+    def sync = null
+    
+    def rethrowIfFailed = null
+    
   }
 }
diff --git a/network/src/test/scala/com/linkedin/norbert/network/netty/ClientStatisticsRequestStrategySpec.scala b/network/src/test/scala/com/linkedin/norbert/network/netty/ClientStatisticsRequestStrategySpec.scala
index 78f71a64..a40df617 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/netty/ClientStatisticsRequestStrategySpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/netty/ClientStatisticsRequestStrategySpec.scala
@@ -35,7 +35,7 @@ class ClientStatisticsRequestStrategySpec extends Specification with Mockito {
       // Give everybody 10 requests
       val requests = nodes.map { node =>
         val uuids = (0 until 10).map(i => UUID.randomUUID)
-        uuids.foreach{ uuid => statsActor.beginRequest(node, uuid) }
+        uuids.foreach{ uuid => statsActor.beginRequest(node, uuid, 0) }
         (node, uuids)
       }.toMap
 
@@ -68,4 +68,4 @@ class ClientStatisticsRequestStrategySpec extends Specification with Mockito {
 
     }
   }
-}
\ No newline at end of file
+}
diff --git a/network/src/test/scala/com/linkedin/norbert/network/server/MessageExecutorSpec.scala b/network/src/test/scala/com/linkedin/norbert/network/server/MessageExecutorSpec.scala
index a310221d..9bc20075 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/server/MessageExecutorSpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/server/MessageExecutorSpec.scala
@@ -40,7 +40,8 @@ class MessageExecutorSpec extends Specification with Mockito with WaitFor with S
     1,
     1,
     100,
-    1000L)
+    1000L,
+    -1)
 
   var handlerCalled = false
   var either: Either[Exception, Ping] = null
diff --git a/network/src/test/scala/com/linkedin/norbert/network/server/NetworkServerSpec.scala b/network/src/test/scala/com/linkedin/norbert/network/server/NetworkServerSpec.scala
index 8ff49bbe..f4dab573 100644
--- a/network/src/test/scala/com/linkedin/norbert/network/server/NetworkServerSpec.scala
+++ b/network/src/test/scala/com/linkedin/norbert/network/server/NetworkServerSpec.scala
@@ -162,7 +162,7 @@ class NetworkServerSpec extends Specification with Mockito with SampleMessage {
 
       got {
         one(networkServer.clusterClient).markNodeAvailable(1, 0)
-        one(networkServer.clusterClient).markNodeUnavailable(1)
+        two(networkServer.clusterClient).markNodeUnavailable(1)
       }
     }
 
diff --git a/norbert/build.gradle b/norbert/build.gradle
new file mode 100644
index 00000000..a71c1417
--- /dev/null
+++ b/norbert/build.gradle
@@ -0,0 +1,18 @@
+apply plugin: 'java'
+
+configurations {
+  compile {
+    transitive = false
+  }
+}
+
+dependencies {
+  compile project(':cluster')
+  compile project(':network')
+  compile project(':java-cluster')
+  compile project(':java-network')
+}
+
+jar {
+    from { configurations.compile.collect { it.isDirectory() ? it : zipTree(it) } }
+}
diff --git a/sbt b/sbt
index 32c584de..17d67f27 100755
--- a/sbt
+++ b/sbt
@@ -1 +1 @@
-java $SBT_OPS -Xms1g -Xmx1g -server -XX:PermSize=128m -XX:MaxPermSize=256m -XX:+UseParallelOldGC -jar `dirname $0`/build/sbt-launch.jar "$@"
+java $SBT_OPS -Xms1g -Xmx1g -server -XX:PermSize=128m -XX:MaxPermSize=256m -XX:+UseParallelOldGC -jar `dirname $0`/lib/sbt-launch.jar "$@"
diff --git a/settings.gradle b/settings.gradle
new file mode 100644
index 00000000..d3537609
--- /dev/null
+++ b/settings.gradle
@@ -0,0 +1,6 @@
+include "cluster"
+include "network"
+include "java-cluster"
+include "java-network"
+include "norbert"
+include "examples"