diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index eee92fa135a..fec21c54e8a 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -70,6 +70,10 @@ Bug Fixes
 
 * SOLR-4330: group.sort is ignored when using group.truncate and ex/tag
   local params together (koji)
+  
+* SOLR-4321: Collections API will sometimes use a node more than once, even
+  when more unused nodes are available. 
+  (Eric Falcao, Brett Hoerner, Mark Miller)
 
 Optimizations
 ----------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java
index f13e6d25aa9..cbde272695f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java
@@ -250,7 +250,7 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
       
       for (int i = 1; i <= numSlices; i++) {
         for (int j = 1; j <= repFactor; j++) {
-          String nodeName = nodeList.get(((i - 1) + (j - 1)) % nodeList.size());
+          String nodeName = nodeList.get((repFactor * (i - 1) + (j - 1)) % nodeList.size());
           String sliceName = "shard" + i;
           String shardName = collectionName + "_" + sliceName + "_replica" + j;
           log.info("Creating shard " + shardName + " as part of slice "
diff --git a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
index 9597af35d06..8ad5e590aca 100644
--- a/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/CollectionsAPIDistributedZkTest.java
@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.net.MalformedURLException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -148,7 +149,8 @@ public class CollectionsAPIDistributedZkTest extends AbstractFullDistribZkTestBa
 
     // would be better if these where all separate tests - but much, much
     // slower
-
+    
+    testNodesUsedByCreate();
     testCollectionsAPI();
 
     // Thread.sleep(10000000000L);
@@ -157,6 +159,51 @@ public class CollectionsAPIDistributedZkTest extends AbstractFullDistribZkTestBa
     }
   }
 
+  private void testNodesUsedByCreate() throws Exception {
+    // we can use this client because we just want base url
+    final String baseUrl = ((HttpSolrServer) clients.get(0)).getBaseURL().substring(
+        0,
+        ((HttpSolrServer) clients.get(0)).getBaseURL().length()
+            - DEFAULT_COLLECTION.length() - 1);
+    
+    ModifiableSolrParams params = new ModifiableSolrParams();
+    params.set("action", CollectionAction.CREATE.toString());
+
+    params.set("numShards", 2);
+    params.set(OverseerCollectionProcessor.REPLICATION_FACTOR, 2);
+    String collectionName = "nodes_used_collection";
+
+    params.set("name", collectionName);
+    QueryRequest request = new QueryRequest(params);
+    request.setPath("/admin/collections");
+    createNewSolrServer("", baseUrl).request(request);
+    
+    List<Integer> numShardsNumReplicaList = new ArrayList<Integer>();
+    numShardsNumReplicaList.add(2);
+    numShardsNumReplicaList.add(2);
+    checkForCollection("nodes_used_collection", numShardsNumReplicaList , null);
+    
+    List<String> createNodeList = new ArrayList<String>();
+
+    Set<String> liveNodes = cloudClient.getZkStateReader().getClusterState()
+        .getLiveNodes();
+    
+    for (String node : liveNodes) {
+      createNodeList.add(node);
+    }
+    
+    DocCollection col = cloudClient.getZkStateReader().getClusterState().getCollection("nodes_used_collection");
+    Collection<Slice> slices = col.getSlices();
+    for (Slice slice : slices) {
+      Collection<Replica> replicas = slice.getReplicas();
+      for (Replica replica : replicas) {
+        createNodeList.remove(replica.getNodeName());
+      }
+    }
+    assertEquals(createNodeList.toString(), 1, createNodeList.size());
+
+  }
+  
   private void testCollectionsAPI() throws Exception {
  
     // TODO: fragile - because we dont pass collection.confName, it will only