You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by an...@apache.org on 2014/05/20 21:48:22 UTC
svn commit: r1596379 - in /lucene/dev/branches/branch_4x: ./ solr/
solr/core/ solr/core/src/java/org/apache/solr/cloud/
solr/core/src/java/org/apache/solr/handler/admin/
solr/core/src/java/org/apache/solr/handler/component/
solr/core/src/test/org/apach...
Author: anshum
Date: Tue May 20 19:48:21 2014
New Revision: 1596379
URL: http://svn.apache.org/r1596379
Log:
SOLR-5681: Make the OverseerCollectionProcessor multi-threaded, merge from trunk (r1596089)
Added:
lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
- copied unchanged from r1596089, lucene/dev/trunk/solr/core/src/test/org/apache/solr/cloud/MultiThreadedOCPTest.java
Modified:
lucene/dev/branches/branch_4x/ (props changed)
lucene/dev/branches/branch_4x/solr/ (props changed)
lucene/dev/branches/branch_4x/solr/CHANGES.txt
lucene/dev/branches/branch_4x/solr/core/ (props changed)
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedMap.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedQueue.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/Overseer.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/ShardHandler.java
lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java
lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerStatusTest.java
lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/core/MockShardHandlerFactory.java
lucene/dev/branches/branch_4x/solr/solrj/ (props changed)
lucene/dev/branches/branch_4x/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
lucene/dev/branches/branch_4x/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
Modified: lucene/dev/branches/branch_4x/solr/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/CHANGES.txt?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/CHANGES.txt (original)
+++ lucene/dev/branches/branch_4x/solr/CHANGES.txt Tue May 20 19:48:21 2014
@@ -92,6 +92,8 @@ Other Changes
recovery scenarios after the leader receives an error trying to forward an
update request to a replica. (Timothy Potter)
+* SOLR-5681: Make the processing of Collection API calls multi-threaded. (Anshum Gupta)
+
Build
---------------------
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedMap.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedMap.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedMap.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedMap.java Tue May 20 19:48:21 2014
@@ -27,6 +27,7 @@ import org.apache.zookeeper.WatchedEvent
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -160,6 +161,12 @@ public class DistributedMap {
return zookeeper.exists(dir + "/" + prefix + trackingId, true);
}
+ public int size() throws KeeperException, InterruptedException {
+ Stat stat = new Stat();
+ zookeeper.getData(dir, null, stat, true);
+ return stat.getNumChildren();
+ }
+
public void remove(String trackingId) throws KeeperException, InterruptedException {
zookeeper.delete(dir + "/" + prefix + trackingId, -1, true);
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedQueue.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedQueue.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedQueue.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/DistributedQueue.java Tue May 20 19:48:21 2014
@@ -18,10 +18,6 @@
package org.apache.solr.cloud;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.TreeMap;
-
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.SolrZkClient;
@@ -36,6 +32,12 @@ import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.TreeMap;
+
/**
* A distributed queue from zk recipes.
*/
@@ -364,6 +366,103 @@ public class DistributedQueue {
}
}
+ public List<QueueEvent> peekTopN(int n, Set<String> excludeSet, Long wait)
+ throws KeeperException, InterruptedException {
+ ArrayList<QueueEvent> topN = new ArrayList<>();
+
+ LOG.debug("Peeking for top {} elements. ExcludeSet: " + excludeSet.toString());
+ boolean waitedEnough = false;
+ TimerContext time = null;
+ if (wait == Long.MAX_VALUE) time = stats.time(dir + "_peekTopN_wait_forever");
+ else time = stats.time(dir + "_peekTopN_wait" + wait);
+
+ try {
+ TreeMap<Long, String> orderedChildren;
+ while (true) {
+ LatchChildWatcher childWatcher = new LatchChildWatcher();
+ try {
+ orderedChildren = orderedChildren(childWatcher);
+ } catch (KeeperException.NoNodeException e) {
+ zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT, true);
+ continue;
+ }
+
+ if (orderedChildren.size() == 0) {
+ if(waitedEnough) return null;
+ childWatcher.await(wait == Long.MAX_VALUE ? DEFAULT_TIMEOUT : wait);
+ waitedEnough = wait != Long.MAX_VALUE;
+ continue;
+ }
+
+ for (String headNode : orderedChildren.values()) {
+ if (headNode != null && topN.size() < n) {
+ try {
+ String id = dir + "/" + headNode;
+ if (excludeSet != null && excludeSet.contains(id)) continue;
+ QueueEvent queueEvent = new QueueEvent(id,
+ zookeeper.getData(dir + "/" + headNode, null, null, true), null);
+ topN.add(queueEvent);
+ } catch (KeeperException.NoNodeException e) {
+ // Another client removed the node first, try next
+ }
+ } else {
+ if (topN.size() >= 1) {
+ printQueueEventsListElementIds(topN);
+ return topN;
+ }
+ }
+ }
+
+ if (topN.size() > 0 ) {
+ printQueueEventsListElementIds(topN);
+ return topN;
+ }
+ if (waitedEnough) {
+ LOG.debug("Waited enough, returning null after peekTopN");
+ return null;
+ }
+ childWatcher.await(wait == Long.MAX_VALUE ? DEFAULT_TIMEOUT : wait);
+ waitedEnough = wait != Long.MAX_VALUE;
+ }
+ } finally {
+ time.stop();
+ }
+ }
+
+ private void printQueueEventsListElementIds(ArrayList<QueueEvent> topN) {
+ if(LOG.isDebugEnabled()) {
+ StringBuffer sb = new StringBuffer("[");
+ for(QueueEvent queueEvent: topN) {
+ sb.append(queueEvent.getId()).append(", ");
+ }
+ sb.append("]");
+ LOG.debug("Returning topN elements: {}", sb.toString());
+ }
+ }
+
+
+ /**
+ *
+ * Gets last element of the Queue without removing it.
+ */
+ public String getTailId() throws KeeperException, InterruptedException {
+ TreeMap<Long, String> orderedChildren = null;
+ orderedChildren = orderedChildren(null);
+ if(orderedChildren == null || orderedChildren.isEmpty()) return null;
+
+ for(String headNode : orderedChildren.descendingMap().values())
+ if (headNode != null) {
+ try {
+ QueueEvent queueEvent = new QueueEvent(dir + "/" + headNode, zookeeper.getData(dir + "/" + headNode,
+ null, null, true), null);
+ return queueEvent.getId();
+ } catch (KeeperException.NoNodeException e) {
+ // Another client removed the node first, try next
+ }
+ }
+ return null;
+ }
+
public static class QueueEvent {
@Override
public int hashCode() {
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/Overseer.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/Overseer.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/Overseer.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/Overseer.java Tue May 20 19:48:21 2014
@@ -1208,7 +1208,6 @@ public class Overseer {
}
}
} finally {
-
if (ccThread != null) {
try {
ccThread.close();
@@ -1317,16 +1316,19 @@ public class Overseer {
public void success(String operation) {
String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
- if (stat == null) {
- stat = new Stat();
- stats.put(op, stat);
+ synchronized (stats) {
+ Stat stat = stats.get(op);
+ if (stat == null) {
+ stat = new Stat();
+ stats.put(op, stat);
+ }
+ stat.success.incrementAndGet();
}
- stat.success.incrementAndGet();
}
public void error(String operation) {
String op = operation.toLowerCase(Locale.ROOT);
+ synchronized (stats) {
Stat stat = stats.get(op);
if (stat == null) {
stat = new Stat();
@@ -1334,20 +1336,26 @@ public class Overseer {
}
stat.errors.incrementAndGet();
}
+ }
public TimerContext time(String operation) {
String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
+ Stat stat;
+ synchronized (stats) {
+ stat = stats.get(op);
if (stat == null) {
stat = new Stat();
stats.put(op, stat);
}
+ }
return stat.requestTime.time();
}
public void storeFailureDetails(String operation, ZkNodeProps request, SolrResponse resp) {
String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
+ Stat stat ;
+ synchronized (stats) {
+ stat = stats.get(op);
if (stat == null) {
stat = new Stat();
stats.put(op, stat);
@@ -1360,6 +1368,7 @@ public class Overseer {
failedOps.addLast(new FailedOp(request, resp));
}
}
+ }
public List<FailedOp> getFailureDetails(String operation) {
Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionProcessor.java Tue May 20 19:48:21 2014
@@ -57,9 +57,11 @@ import org.apache.solr.common.util.Simpl
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.handler.component.HttpShardHandlerFactory;
import org.apache.solr.handler.component.ShardHandler;
+import org.apache.solr.handler.component.ShardHandlerFactory;
import org.apache.solr.handler.component.ShardRequest;
import org.apache.solr.handler.component.ShardResponse;
import org.apache.solr.update.SolrIndexSplitter;
+import org.apache.solr.util.DefaultSolrThreadFactory;
import org.apache.solr.util.stats.Snapshot;
import org.apache.solr.util.stats.Timer;
import org.apache.solr.util.stats.TimerContext;
@@ -83,6 +85,8 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -123,8 +127,6 @@ public class OverseerCollectionProcessor
public static final String DELETESHARD = "deleteshard";
- public static final String REQUESTSTATUS = "status";
-
public static final String ROUTER = "router";
public static final String SHARDS_PROP = "shards";
@@ -143,6 +145,8 @@ public class OverseerCollectionProcessor
public static final String COLL_PROP_PREFIX = "property.";
+ public int maxParallelThreads = 10;
+
public static final Set<String> KNOWN_CLUSTER_PROPS = ImmutableSet.of(ZkStateReader.LEGACY_CLOUD, ZkStateReader.URL_SCHEME);
public static final Map<String,Object> COLL_PROPS = ZkNodeProps.makeMap(
@@ -152,8 +156,7 @@ public class OverseerCollectionProcessor
"external",null );
- // TODO: use from Overseer?
- private static final String QUEUE_OPERATION = "operation";
+ public ExecutorService tpe ;
private static Logger log = LoggerFactory
.getLogger(OverseerCollectionProcessor.class);
@@ -162,10 +165,20 @@ public class OverseerCollectionProcessor
private DistributedMap runningMap;
private DistributedMap completedMap;
private DistributedMap failureMap;
-
+
+ // Set that maintains a list of all the tasks that are running. This is keyed on zk id of the task.
+ final private Set runningTasks;
+
+ // Set that tracks collections that are currently being processed by a running task.
+ // This is used for handling mutual exclusion of the tasks.
+ final private Set collectionWip;
+
+ // List of completed tasks. This is used to clean up workQueue in zk.
+ final private HashMap<String, QueueEvent> completedTasks;
+
private String myId;
- private ShardHandler shardHandler;
+ private final ShardHandlerFactory shardHandlerFactory;
private String adminPath;
@@ -175,13 +188,22 @@ public class OverseerCollectionProcessor
private Overseer.Stats stats;
- public OverseerCollectionProcessor(ZkStateReader zkStateReader, String myId, ShardHandler shardHandler, String adminPath, Overseer.Stats stats) {
- this(zkStateReader, myId, shardHandler, adminPath, stats, Overseer.getCollectionQueue(zkStateReader.getZkClient(), stats),
+ // Set of tasks that have been picked up for processing but not cleaned up from zk work-queue.
+ // It may contain tasks that have completed execution, have been entered into the completed/failed map in zk but not
+ // deleted from the work-queue as that is a batched operation.
+ final private Set<String> runningZKTasks;
+ private final Object waitLock = new Object();
+
+ public OverseerCollectionProcessor(ZkStateReader zkStateReader, String myId,
+ final ShardHandler shardHandler,
+ String adminPath, Overseer.Stats stats) {
+ this(zkStateReader, myId, shardHandler.getShardHandlerFactory(), adminPath, stats, Overseer.getCollectionQueue(zkStateReader.getZkClient(), stats),
Overseer.getRunningMap(zkStateReader.getZkClient()),
Overseer.getCompletedMap(zkStateReader.getZkClient()), Overseer.getFailureMap(zkStateReader.getZkClient()));
}
- protected OverseerCollectionProcessor(ZkStateReader zkStateReader, String myId, ShardHandler shardHandler,
+ protected OverseerCollectionProcessor(ZkStateReader zkStateReader, String myId,
+ final ShardHandlerFactory shardHandlerFactory,
String adminPath,
Overseer.Stats stats,
DistributedQueue workQueue,
@@ -190,107 +212,199 @@ public class OverseerCollectionProcessor
DistributedMap failureMap) {
this.zkStateReader = zkStateReader;
this.myId = myId;
- this.shardHandler = shardHandler;
+ this.shardHandlerFactory = shardHandlerFactory;
this.adminPath = adminPath;
this.workQueue = workQueue;
this.runningMap = runningMap;
this.completedMap = completedMap;
this.failureMap = failureMap;
this.stats = stats;
+ this.runningZKTasks = new HashSet<>();
+ this.runningTasks = new HashSet();
+ this.collectionWip = new HashSet();
+ this.completedTasks = new HashMap<>();
}
@Override
public void run() {
- log.info("Process current queue of collection creations");
- LeaderStatus isLeader = amILeader();
- while (isLeader == LeaderStatus.DONT_KNOW) {
- log.debug("am_i_leader unclear {}", isLeader);
- isLeader = amILeader(); // not a no, not a yes, try ask again
- }
+ log.info("Process current queue of collection creations");
+ LeaderStatus isLeader = amILeader();
+ while (isLeader == LeaderStatus.DONT_KNOW) {
+ log.debug("am_i_leader unclear {}", isLeader);
+ isLeader = amILeader(); // not a no, not a yes, try ask again
+ }
+
+ String oldestItemInWorkQueue = null;
+ // hasLeftOverItems - used for avoiding re-execution of async tasks that were processed by a previous Overseer.
+ // This variable is set in case there's any task found on the workQueue when the OCP starts up and
+ // the id for the queue tail is used as a marker to check for the task in completed/failed map in zk.
+ // Beyond the marker, all tasks can safely be assumed to have never been executed.
+ boolean hasLeftOverItems = true;
+
+ try {
+ oldestItemInWorkQueue = workQueue.getTailId();
+ } catch (KeeperException e) {
+ // We don't need to handle this. This is just a fail-safe which comes in handy in skipping already processed
+ // async calls.
+ SolrException.log(log, "", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ if (oldestItemInWorkQueue == null)
+ hasLeftOverItems = false;
+ else
+ log.debug("Found already existing elements in the work-queue. Last element: {}", oldestItemInWorkQueue);
try {
prioritizeOverseerNodes();
} catch (Exception e) {
log.error("Unable to prioritize overseer ", e);
-
}
- while (!this.isClosed) {
- try {
- isLeader = amILeader();
- if (LeaderStatus.NO == isLeader) {
- break;
- }
- else if (LeaderStatus.YES != isLeader) {
- log.debug("am_i_leader unclear {}", isLeader);
- continue; // not a no, not a yes, try asking again
- }
-
- QueueEvent head = workQueue.peek(true);
- if(isClosed) break;
- final ZkNodeProps message = ZkNodeProps.load(head.getBytes());
-
- final String asyncId = (message.containsKey(ASYNC) && message.get(ASYNC) != null) ? (String) message.get(ASYNC) : null;
-
- try {
- if(message.containsKey(ASYNC) && message.get(ASYNC) != null && !runningMap.contains(message.getStr(ASYNC)))
- runningMap.put(asyncId, null);
- } catch (KeeperException.NodeExistsException e) {
- // Just catch and do nothing. The runningMap.contains(..) check ensures that this is the only
- // entry point into the runningMap.
- // NOTE: Make sure to handle it as soon as OCP gets distributed/multi-threaded.
- }
-
- log.info("Overseer Collection Processor: Get the message id:" + head.getId() + " message:" + message.toString());
- final String operation = message.getStr(QUEUE_OPERATION);
- final TimerContext timerContext = stats.time("collection_" + operation); // even if operation is async, it is sync!
- SolrResponse response = null;
- try {
- response = processMessage(message, operation);
- } finally {
- timerContext.stop();
- }
-
- head.setBytes(SolrResponse.serializable(response));
- if (!operation.equals(REQUESTSTATUS) && asyncId != null) {
- if(response.getResponse().get("failure") != null || response.getResponse().get("exception") != null) {
- failureMap.put(asyncId, null);
- } else {
- completedMap.put(asyncId, null);
- }
- }
- if(asyncId != null)
- runningMap.remove(asyncId);
-
- workQueue.remove(head);
-
- if (response.getResponse().get("failure") != null || response.getResponse().get("exception") != null) {
- stats.error("collection_" + operation);
- stats.storeFailureDetails("collection_" + operation, message, response);
- } else {
- stats.success("collection_" + operation);
- }
- log.info("Overseer Collection Processor: Message id:" + head.getId() + " complete, response:"+ response.getResponse().toString());
+ // TODO: Make maxThreads configurable.
+
+ this.tpe = new ThreadPoolExecutor(5, 100, 0L, TimeUnit.MILLISECONDS,
+ new SynchronousQueue<Runnable>(),
+ new DefaultSolrThreadFactory("OverseerThreadFactory"));
+ try {
+ while (!this.isClosed) {
+ try {
+ isLeader = amILeader();
+ if (LeaderStatus.NO == isLeader) {
+ break;
+ } else if (LeaderStatus.YES != isLeader) {
+ log.debug("am_i_leader unclear {}", isLeader);
+ continue; // not a no, not a yes, try asking again
+ }
+
+ log.debug("Cleaning up work-queue. #Running tasks: {}", runningTasks.size());
+ cleanUpWorkQueue();
+
+ printTrackingMaps();
+
+ boolean waited = false;
+
+ while (runningTasks.size() > maxParallelThreads) {
+ synchronized (waitLock) {
+ waitLock.wait(100);//wait for 100 ms or till a task is complete
+ }
+ waited = true;
+ }
+
+ if (waited)
+ cleanUpWorkQueue();
+
+ List<QueueEvent> heads = workQueue.peekTopN(maxParallelThreads, runningZKTasks, 2000L);
+
+ if (heads == null)
+ continue;
+
+ log.debug("Got {} tasks from work-queue : [{}]", heads.size(), heads.toString());
+
+ if (isClosed) break;
+
+ for (QueueEvent head : heads) {
+ final ZkNodeProps message = ZkNodeProps.load(head.getBytes());
+ String collectionName = message.containsKey(COLLECTION_PROP) ?
+ message.getStr(COLLECTION_PROP) : message.getStr("name");
+ String asyncId = message.getStr(ASYNC);
+ if (hasLeftOverItems) {
+ if (head.getId().equals(oldestItemInWorkQueue))
+ hasLeftOverItems = false;
+ if (asyncId != null && (completedMap.contains(asyncId) || failureMap.contains(asyncId))) {
+ log.debug("Found already processed task in workQueue, cleaning up. AsyncId [{}]",asyncId );
+ workQueue.remove(head);
+ continue;
+ }
+ }
+
+ if (!checkExclusivity(message, head.getId())) {
+ log.debug("Exclusivity check failed for [{}]", message.toString());
+ continue;
+ }
+
+ try {
+ markTaskAsRunning(head, collectionName, asyncId);
+ log.debug("Marked task [{}] as running", head.getId());
+ } catch (KeeperException.NodeExistsException e) {
+ // This should never happen
+ log.error("Tried to pick up task [{}] when it was already running!", head.getId());
+ } catch (InterruptedException e) {
+ log.error("Thread interrupted while trying to pick task for execution.", head.getId());
+ Thread.currentThread().interrupt();
+ }
+
+ log.info("Overseer Collection Processor: Get the message id:" + head.getId() + " message:" + message.toString());
+ String operation = message.getStr(Overseer.QUEUE_OPERATION);
+ Runner runner = new Runner(message,
+ operation, head);
+ tpe.execute(runner);
+ }
+
} catch (KeeperException e) {
if (e.code() == KeeperException.Code.SESSIONEXPIRED
|| e.code() == KeeperException.Code.CONNECTIONLOSS) {
- log.warn("Overseer cannot talk to ZK");
- return;
- }
- SolrException.log(log, "", e);
- throw new ZooKeeperException(
- SolrException.ErrorCode.SERVER_ERROR, "", e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return;
- } catch (Exception e) {
- SolrException.log(log, "", e);
- }
- }
+ log.warn("Overseer cannot talk to ZK");
+ return;
+ }
+ SolrException.log(log, "", e);
+ throw new ZooKeeperException(
+ SolrException.ErrorCode.SERVER_ERROR, "", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ } catch (Exception e) {
+ SolrException.log(log, "", e);
+ }
+ }
+ } finally {
+ this.close();
+ }
}
-
+
+ private boolean checkExclusivity(ZkNodeProps message, String id) throws KeeperException, InterruptedException {
+ String collectionName = message.containsKey(COLLECTION_PROP) ?
+ message.getStr(COLLECTION_PROP) : message.getStr("name");
+
+ if(collectionName == null)
+ return true;
+
+ if(collectionWip.contains(collectionName))
+ return false;
+
+ if(runningZKTasks.contains(id))
+ return false;
+
+ return true;
+ }
+
+ private void cleanUpWorkQueue() throws KeeperException, InterruptedException {
+ synchronized (completedTasks) {
+ for (String id : completedTasks.keySet()) {
+ workQueue.remove(completedTasks.get(id));
+ runningZKTasks.remove(id);
+ }
+ completedTasks.clear();
+ }
+ }
+
public void close() {
isClosed = true;
+ if(tpe != null) {
+ if (!tpe.isShutdown()) {
+ tpe.shutdown();
+ try {
+ tpe.awaitTermination(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ log.warn("Thread interrupted while waiting for OCP threadpool shutdown.");
+ Thread.currentThread().interrupt();
+ } finally {
+ if (!tpe.isShutdown())
+ tpe.shutdownNow();
+ }
+ }
+ }
}
private void prioritizeOverseerNodes() throws KeeperException, InterruptedException {
@@ -308,9 +422,8 @@ public class OverseerCollectionProcessor
if(nodeNames.size()<2) return;
boolean designateIsInFront = overseerDesignates.contains( nodeNames.get(0));
-//
ArrayList<String> nodesTobePushedBack = new ArrayList<>();
- //ensure that the node right behind the leader , i.r at position 1 is a Overseer
+ //ensure that the node right behind the leader , i.e at position 1 is a Overseer
List<String> availableDesignates = new ArrayList<>();
log.info("sorted nodes {}", nodeNames);//TODO to be removed
@@ -414,6 +527,7 @@ public class OverseerCollectionProcessor
private void invokeOverseerOp(String nodeName, String op) {
ModifiableSolrParams params = new ModifiableSolrParams();
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
params.set(CoreAdminParams.ACTION, CoreAdminAction.OVERSEEROP.toString());
params.set("op", op);
params.set("qt", adminPath);
@@ -449,8 +563,8 @@ public class OverseerCollectionProcessor
log.info("According to ZK I (id=" + myId + ") am no longer a leader.");
return LeaderStatus.NO;
}
-
-
+
+
protected SolrResponse processMessage(ZkNodeProps message, String operation) {
log.warn("OverseerCollectionProcessor.processMessage : "+ operation + " , "+ message.toString());
@@ -487,12 +601,11 @@ public class OverseerCollectionProcessor
} else if(LIST.isEqual(operation)) {
listCollections(zkStateReader.getClusterState(), results);
} else if (CLUSTERSTATUS.isEqual(operation)) {
- getClusterStatus(zkStateReader.getClusterState(), message, results);
+ getClusterStatus(zkStateReader.getClusterState(), message, results);
} else {
throw new SolrException(ErrorCode.BAD_REQUEST, "Unknown operation:"
+ operation);
}
-
} catch (Exception e) {
SolrException.log(log, "Collection " + operation + " of " + operation
+ " failed", e);
@@ -501,8 +614,7 @@ public class OverseerCollectionProcessor
nl.add("msg", e.getMessage());
nl.add("rspCode", e instanceof SolrException ? ((SolrException)e).code() : -1);
results.add("exception", nl);
- }
-
+ }
return new OverseerSolrResponse(results);
}
@@ -524,16 +636,16 @@ public class OverseerCollectionProcessor
NamedList stateUpdateQueueStats = new NamedList();
NamedList workQueueStats = new NamedList();
NamedList collectionQueueStats = new NamedList();
- for (Map.Entry<String, Overseer.Stat> entry : this.stats.getStats().entrySet()) {
+ for (Map.Entry<String, Overseer.Stat> entry : stats.getStats().entrySet()) {
String key = entry.getKey();
NamedList<Object> lst = new SimpleOrderedMap<>();
if (key.startsWith("collection_")) {
collectionStats.add(key.substring(11), lst);
- int successes = this.stats.getSuccessCount(entry.getKey());
- int errors = this.stats.getErrorCount(entry.getKey());
+ int successes = stats.getSuccessCount(entry.getKey());
+ int errors = stats.getErrorCount(entry.getKey());
lst.add("requests", successes);
lst.add("errors", errors);
- List<Overseer.FailedOp> failureDetails = this.stats.getFailureDetails(key);
+ List<Overseer.FailedOp> failureDetails = stats.getFailureDetails(key);
if (failureDetails != null) {
List<SimpleOrderedMap<Object>> failures = new ArrayList<>();
for (Overseer.FailedOp failedOp : failureDetails) {
@@ -553,8 +665,8 @@ public class OverseerCollectionProcessor
} else {
// overseer stats
overseerStats.add(key, lst);
- int successes = this.stats.getSuccessCount(entry.getKey());
- int errors = this.stats.getErrorCount(entry.getKey());
+ int successes = stats.getSuccessCount(entry.getKey());
+ int errors = stats.getErrorCount(entry.getKey());
lst.add("requests", successes);
lst.add("errors", errors);
}
@@ -754,6 +866,7 @@ public class OverseerCollectionProcessor
String replicaName = message.getStr(REPLICA_PROP);
DocCollection coll = clusterState.getCollection(collectionName);
Slice slice = coll.getSlice(shard);
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
if(slice==null){
throw new SolrException(ErrorCode.BAD_REQUEST, "Invalid shard name : "+shard+" in collection : "+ collectionName);
}
@@ -783,7 +896,8 @@ public class OverseerCollectionProcessor
log.warn("Exception trying to unload core " + sreq, e);
}
- collectShardResponses(!Slice.ACTIVE.equals(replica.getStr(Slice.STATE)) ? new NamedList() : results, false, null);
+ collectShardResponses(!Slice.ACTIVE.equals(replica.getStr(Slice.STATE)) ? new NamedList() : results,
+ false, null, shardHandler);
if (waitForCoreNodeGone(collectionName, shard, replicaName, 5000)) return;//check if the core unload removed the corenode zk enry
deleteCoreNode(collectionName, replicaName, replica, core); // try and ensure core info is removed from clusterstate
@@ -976,7 +1090,8 @@ public class OverseerCollectionProcessor
}
- private boolean createShard(ClusterState clusterState, ZkNodeProps message, NamedList results) throws KeeperException, InterruptedException {
+ private boolean createShard(ClusterState clusterState, ZkNodeProps message, NamedList results)
+ throws KeeperException, InterruptedException {
log.info("Create shard invoked: {}", message);
String collectionName = message.getStr(COLLECTION_PROP);
String shard = message.getStr(SHARD_ID_PROP);
@@ -984,6 +1099,7 @@ public class OverseerCollectionProcessor
throw new SolrException(ErrorCode.BAD_REQUEST, "'collection' and 'shard' are required parameters" );
int numSlices = 1;
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
DocCollection collection = clusterState.getCollection(collectionName);
int maxShardsPerNode = collection.getInt(MAX_SHARDS_PER_NODE, 1);
int repFactor = message.getInt(REPLICATION_FACTOR, collection.getInt(REPLICATION_FACTOR, 1));
@@ -1036,7 +1152,7 @@ public class OverseerCollectionProcessor
}
- processResponses(results);
+ processResponses(results, shardHandler);
log.info("Finished create command on all shards for collection: "
+ collectionName);
@@ -1050,6 +1166,7 @@ public class OverseerCollectionProcessor
String collectionName = message.getStr("collection");
String slice = message.getStr(ZkStateReader.SHARD_ID_PROP);
String splitKey = message.getStr("split.key");
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
DocCollection collection = clusterState.getCollection(collectionName);
DocRouter router = collection.getRouter() != null ? collection.getRouter() : DocRouter.DEFAULT;
@@ -1195,7 +1312,7 @@ public class OverseerCollectionProcessor
// do not abort splitshard if the unloading fails
// this can happen because the replicas created previously may be down
// the only side effect of this is that the sub shard may end up having more replicas than we want
- collectShardResponses(results, false, null);
+ collectShardResponses(results, false, null, shardHandler);
String asyncId = message.getStr(ASYNC);
HashMap<String, String> requestMap = new HashMap<String, String>();
@@ -1228,11 +1345,11 @@ public class OverseerCollectionProcessor
params.set(CoreAdminParams.SHARD, subSlice);
setupAsyncRequest(asyncId, requestMap, params, nodeName);
addPropertyParams(message, params);
- sendShardRequest(nodeName, params);
+ sendShardRequest(nodeName, params, shardHandler);
}
collectShardResponses(results, true,
- "SPLITSHARD failed to create subshard leaders");
+ "SPLITSHARD failed to create subshard leaders", shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1251,11 +1368,11 @@ public class OverseerCollectionProcessor
ModifiableSolrParams p = new ModifiableSolrParams(cmd.getParams());
setupAsyncRequest(asyncId, requestMap, p, nodeName);
- sendShardRequest(nodeName, p);
+ sendShardRequest(nodeName, p, shardHandler);
}
collectShardResponses(results, true,
- "SPLITSHARD timed out waiting for subshard leaders to come up");
+ "SPLITSHARD timed out waiting for subshard leaders to come up", shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1276,9 +1393,10 @@ public class OverseerCollectionProcessor
params.set(CoreAdminParams.RANGES, rangesStr);
setupAsyncRequest(asyncId, requestMap, params, parentShardLeader.getNodeName());
- sendShardRequest(parentShardLeader.getNodeName(), params);
+ sendShardRequest(parentShardLeader.getNodeName(), params, shardHandler);
- collectShardResponses(results, true, "SPLITSHARD failed to invoke SPLIT core admin command");
+ collectShardResponses(results, true, "SPLITSHARD failed to invoke SPLIT core admin command",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
log.info("Index on shard: " + nodeName + " split into two successfully");
@@ -1295,11 +1413,12 @@ public class OverseerCollectionProcessor
setupAsyncRequest(asyncId, requestMap, params, nodeName);
- sendShardRequest(nodeName, params);
+ sendShardRequest(nodeName, params, shardHandler);
}
collectShardResponses(results, true,
- "SPLITSHARD failed while asking sub shard leaders to apply buffered updates");
+ "SPLITSHARD failed while asking sub shard leaders to apply buffered updates",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1365,7 +1484,7 @@ public class OverseerCollectionProcessor
//Not using this property. Do we really need to use it?
//params.set(ZkStateReader.NUM_SHARDS_PROP, numSlices);
- sendShardRequest(subShardNodeName, params);
+ sendShardRequest(subShardNodeName, params, shardHandler);
String coreNodeName = waitForCoreNodeName(collection, subShardNodeName, shardName);
// wait for the replicas to be seen as active on sub shard leader
@@ -1381,13 +1500,14 @@ public class OverseerCollectionProcessor
setupAsyncRequest(asyncId, requestMap, p, nodeName);
- sendShardRequest(nodeName, p);
+ sendShardRequest(nodeName, p, shardHandler);
}
}
collectShardResponses(results, true,
- "SPLITSHARD failed to create subshard replicas or timed out waiting for them to come up");
+ "SPLITSHARD failed to create subshard replicas or timed out waiting for them to come up",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1445,6 +1565,7 @@ public class OverseerCollectionProcessor
}
}
+
static UpdateResponse softCommit(String url) throws SolrServerException, IOException {
HttpSolrServer server = null;
try {
@@ -1491,7 +1612,9 @@ public class OverseerCollectionProcessor
throw new SolrException(ErrorCode.SERVER_ERROR, "Could not find coreNodeName");
}
- private void collectShardResponses(NamedList results, boolean abortOnError, String msgOnError) {
+ private void collectShardResponses(NamedList results, boolean abortOnError,
+ String msgOnError,
+ ShardHandler shardHandler) {
ShardResponse srsp;
do {
srsp = shardHandler.takeCompletedOrError();
@@ -1532,14 +1655,15 @@ public class OverseerCollectionProcessor
"The slice: " + slice.getName() + " is currently "
+ slice.getState() + ". Only INACTIVE (or custom-hashed) slices can be deleted.");
}
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
try {
ModifiableSolrParams params = new ModifiableSolrParams();
params.set(CoreAdminParams.ACTION, CoreAdminAction.UNLOAD.toString());
params.set(CoreAdminParams.DELETE_INDEX, "true");
- sliceCmd(clusterState, params, null, slice);
+ sliceCmd(clusterState, params, null, slice, shardHandler);
- processResponses(results);
+ processResponses(results, shardHandler);
ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION,
Overseer.REMOVESHARD, ZkStateReader.COLLECTION_PROP, collection,
@@ -1625,7 +1749,7 @@ public class OverseerCollectionProcessor
if (clusterState.hasCollection(tempSourceCollectionName)) {
log.info("Deleting temporary collection: " + tempSourceCollectionName);
Map<String, Object> props = ZkNodeProps.makeMap(
- QUEUE_OPERATION, DELETECOLLECTION,
+ Overseer.QUEUE_OPERATION, DELETECOLLECTION,
"name", tempSourceCollectionName);
try {
@@ -1638,6 +1762,8 @@ public class OverseerCollectionProcessor
CompositeIdRouter sourceRouter = (CompositeIdRouter) sourceCollection.getRouter();
DocRouter.Range keyHashRange = sourceRouter.keyHashRange(splitKey);
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
+
log.info("Hash range for split.key: {} is: {}", splitKey, keyHashRange);
// intersect source range, keyHashRange and target range
// this is the range that has to be split from source and transferred to target
@@ -1660,9 +1786,10 @@ public class OverseerCollectionProcessor
String nodeName = targetLeader.getNodeName();
setupAsyncRequest(asyncId, requestMap, params, nodeName);
- sendShardRequest(targetLeader.getNodeName(), params);
+ sendShardRequest(targetLeader.getNodeName(), params, shardHandler);
- collectShardResponses(results, true, "MIGRATE failed to request node to buffer updates");
+ collectShardResponses(results, true, "MIGRATE failed to request node to buffer updates",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1706,7 +1833,7 @@ public class OverseerCollectionProcessor
// create a temporary collection with just one node on the shard leader
String configName = zkStateReader.readConfigName(sourceCollection.getName());
Map<String, Object> props = ZkNodeProps.makeMap(
- QUEUE_OPERATION, CREATECOLLECTION,
+ Overseer.QUEUE_OPERATION, CREATECOLLECTION,
"name", tempSourceCollectionName,
REPLICATION_FACTOR, 1,
NUM_SLICES, 1,
@@ -1736,10 +1863,11 @@ public class OverseerCollectionProcessor
cmd.setState(ZkStateReader.ACTIVE);
cmd.setCheckLive(true);
cmd.setOnlyIfLeader(true);
- sendShardRequest(tempSourceLeader.getNodeName(), new ModifiableSolrParams(cmd.getParams()));
+ sendShardRequest(tempSourceLeader.getNodeName(), new ModifiableSolrParams(cmd.getParams()), shardHandler);
collectShardResponses(results, true,
- "MIGRATE failed to create temp collection leader or timed out waiting for it to come up");
+ "MIGRATE failed to create temp collection leader or timed out waiting for it to come up",
+ shardHandler);
log.info("Asking source leader to split index");
params = new ModifiableSolrParams();
@@ -1753,8 +1881,8 @@ public class OverseerCollectionProcessor
setupAsyncRequest(asyncId, requestMap, params, tempNodeName);
- sendShardRequest(tempNodeName, params);
- collectShardResponses(results, true, "MIGRATE failed to invoke SPLIT core admin command");
+ sendShardRequest(tempNodeName, params, shardHandler);
+ collectShardResponses(results, true, "MIGRATE failed to invoke SPLIT core admin command", shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
log.info("Creating a replica of temporary collection: {} on the target leader node: {}",
@@ -1767,9 +1895,10 @@ public class OverseerCollectionProcessor
params.set(CoreAdminParams.SHARD, tempSourceSlice.getName());
setupAsyncRequest(asyncId, requestMap, params, targetLeader.getNodeName());
- sendShardRequest(targetLeader.getNodeName(), params);
+ sendShardRequest(targetLeader.getNodeName(), params, shardHandler);
collectShardResponses(results, true,
- "MIGRATE failed to create replica of temporary collection in target leader node.");
+ "MIGRATE failed to create replica of temporary collection in target leader node.",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1788,10 +1917,11 @@ public class OverseerCollectionProcessor
setupAsyncRequest(asyncId, requestMap, params, tempSourceLeader.getNodeName());
- sendShardRequest(tempSourceLeader.getNodeName(), params);
+ sendShardRequest(tempSourceLeader.getNodeName(), params, shardHandler);
collectShardResponses(results, true,
- "MIGRATE failed to create temp collection replica or timed out waiting for them to come up");
+ "MIGRATE failed to create temp collection replica or timed out waiting for them to come up",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
log.info("Successfully created replica of temp source collection on target leader node");
@@ -1804,9 +1934,11 @@ public class OverseerCollectionProcessor
setupAsyncRequest(asyncId, requestMap, params, sourceLeader.getNodeName());
- sendShardRequest(targetLeader.getNodeName(), params);
+ sendShardRequest(targetLeader.getNodeName(), params, shardHandler);
collectShardResponses(results, true,
- "MIGRATE failed to merge " + tempCollectionReplica2 + " to " + targetLeader.getStr("core") + " on node: " + targetLeader.getNodeName());
+ "MIGRATE failed to merge " + tempCollectionReplica2 +
+ " to " + targetLeader.getStr("core") + " on node: " + targetLeader.getNodeName(),
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
@@ -1816,16 +1948,17 @@ public class OverseerCollectionProcessor
params.set(CoreAdminParams.NAME, targetLeader.getStr("core"));
setupAsyncRequest(asyncId, requestMap, params, targetLeader.getNodeName());
- sendShardRequest(targetLeader.getNodeName(), params);
+ sendShardRequest(targetLeader.getNodeName(), params, shardHandler);
collectShardResponses(results, true,
- "MIGRATE failed to request node to apply buffered updates");
+ "MIGRATE failed to request node to apply buffered updates",
+ shardHandler);
completeAsyncRequest(asyncId, requestMap, results);
try {
log.info("Deleting temporary collection: " + tempSourceCollectionName);
props = ZkNodeProps.makeMap(
- QUEUE_OPERATION, DELETECOLLECTION,
+ Overseer.QUEUE_OPERATION, DELETECOLLECTION,
"name", tempSourceCollectionName);
deleteCollection(new ZkNodeProps(props), results);
} catch (Exception e) {
@@ -1863,7 +1996,7 @@ public class OverseerCollectionProcessor
}
}
- private void sendShardRequest(String nodeName, ModifiableSolrParams params) {
+ private void sendShardRequest(String nodeName, ModifiableSolrParams params, ShardHandler shardHandler) {
ShardRequest sreq = new ShardRequest();
params.set("qt", adminPath);
sreq.purpose = 1;
@@ -1883,6 +2016,7 @@ public class OverseerCollectionProcessor
}
}
}
+
private void createCollection(ClusterState clusterState, ZkNodeProps message, NamedList results) throws KeeperException, InterruptedException {
String collectionName = message.getStr("name");
if (clusterState.hasCollection(collectionName)) {
@@ -1895,9 +2029,9 @@ public class OverseerCollectionProcessor
int repFactor = message.getInt( REPLICATION_FACTOR, 1);
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
String async = null;
- if (message.containsKey("async"))
- async = message.getStr("async");
+ async = message.getStr("async");
Integer numSlices = message.getInt(NUM_SLICES, null);
String router = message.getStr("router.name", DocRouter.DEFAULT_NAME);
@@ -1987,7 +2121,8 @@ public class OverseerCollectionProcessor
// For tracking async calls.
HashMap<String, String> requestMap = new HashMap<String, String>();
- log.info("Creating SolrCores for new collection, shardNames {} , replicationFactor : {}", shardNames, repFactor);
+ log.info("Creating SolrCores for new collection {}, shardNames {} , replicationFactor : {}",
+ collectionName, shardNames, repFactor);
Map<String ,ShardRequest> coresToCreate = new LinkedHashMap<>();
for (int i = 1; i <= shardNames.size(); i++) {
String sliceName = shardNames.get(i-1);
@@ -2052,7 +2187,7 @@ public class OverseerCollectionProcessor
}
}
- processResponses(results);
+ processResponses(results, shardHandler);
completeAsyncRequest(async, requestMap, results);
@@ -2110,6 +2245,7 @@ public class OverseerCollectionProcessor
throw new SolrException(ErrorCode.BAD_REQUEST,
"Collection: " + collection + " shard: " + shard + " does not exist");
}
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
if(node== null){
node = getNodesForNewShard(clusterState,collection, coll.getSlices().size() , coll.getInt(MAX_SHARDS_PER_NODE, 1),coll.getInt(REPLICATION_FACTOR, 1),null).get(0).nodeName;
@@ -2182,14 +2318,13 @@ public class OverseerCollectionProcessor
}
addPropertyParams(message, params);
- sendShardRequest(node, params);
+ sendShardRequest(node, params, shardHandler);
collectShardResponses(results, true,
- "ADDREPLICA failed to create replica");
+ "ADDREPLICA failed to create replica", shardHandler);
}
-
- private void processResponses(NamedList results) {
+ private void processResponses(NamedList results, ShardHandler shardHandler) {
ShardResponse srsp;
do {
srsp = shardHandler.takeCompletedOrError();
@@ -2199,7 +2334,6 @@ public class OverseerCollectionProcessor
} while (srsp != null);
}
-
private String createConfNode(String coll, ZkNodeProps message, boolean isLegacyCloud) throws KeeperException, InterruptedException {
String configName = message.getStr(OverseerCollectionProcessor.COLL_CONF);
if(configName == null){
@@ -2237,19 +2371,21 @@ public class OverseerCollectionProcessor
private void collectionCmd(ClusterState clusterState, ZkNodeProps message, ModifiableSolrParams params, NamedList results, String stateMatcher) {
log.info("Executing Collection Cmd : " + params);
String collectionName = message.getStr("name");
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
DocCollection coll = clusterState.getCollection(collectionName);
for (Map.Entry<String,Slice> entry : coll.getSlicesMap().entrySet()) {
Slice slice = entry.getValue();
- sliceCmd(clusterState, params, stateMatcher, slice);
+ sliceCmd(clusterState, params, stateMatcher, slice, shardHandler);
}
- processResponses(results);
+ processResponses(results, shardHandler);
}
- private void sliceCmd(ClusterState clusterState, ModifiableSolrParams params, String stateMatcher, Slice slice) {
+ private void sliceCmd(ClusterState clusterState, ModifiableSolrParams params, String stateMatcher,
+ Slice slice, ShardHandler shardHandler) {
Map<String,Replica> shards = slice.getReplicasMap();
Set<Map.Entry<String,Replica>> shardEntries = shards.entrySet();
for (Map.Entry<String,Replica> shardEntry : shardEntries) {
@@ -2317,12 +2453,13 @@ public class OverseerCollectionProcessor
private void waitForAsyncCallsToComplete(Map<String, String> requestMap, NamedList results) {
for(String k:requestMap.keySet()) {
- log.debug("I am Waiting for : " + k + "/" + requestMap.get(k));
+ log.debug("I am Waiting for :{}/{}", k, requestMap.get(k));
results.add(requestMap.get(k), waitForCoreAdminAsyncCallToComplete(k, requestMap.get(k)));
}
}
private NamedList waitForCoreAdminAsyncCallToComplete(String nodeName, String requestId) {
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
ModifiableSolrParams params = new ModifiableSolrParams();
params.set(CoreAdminParams.ACTION, CoreAdminAction.REQUESTSTATUS.toString());
params.set(CoreAdminParams.REQUESTID, requestId);
@@ -2351,7 +2488,7 @@ public class OverseerCollectionProcessor
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
- e.printStackTrace();
+ Thread.currentThread().interrupt();
}
continue;
@@ -2380,6 +2517,162 @@ public class OverseerCollectionProcessor
} while (srsp != null);
} while(true);
}
+
+ private void markTaskAsRunning(QueueEvent head, String collectionName,String asyncId)
+ throws KeeperException, InterruptedException {
+ synchronized (runningZKTasks) {
+ runningZKTasks.add(head.getId());
+ }
+
+ synchronized (runningTasks) {
+ runningTasks.add(head.getId());
+ }
+ if(collectionName != null) {
+ synchronized (collectionWip) {
+ collectionWip.add(collectionName);
+ }
+ }
+
+ if(asyncId != null)
+ runningMap.put(asyncId, null);
+ }
+
+ protected class Runner implements Runnable {
+ ZkNodeProps message;
+ String operation;
+ SolrResponse response;
+ QueueEvent head;
+
+ public Runner(ZkNodeProps message, String operation, QueueEvent head) {
+ this.message = message;
+ this.operation = operation;
+ this.head = head;
+ response = null;
+ }
+
+
+ @Override
+ public void run() {
+
+ final TimerContext timerContext = stats.time("collection_" + operation);
+
+ boolean success = false;
+ String asyncId = message.getStr(ASYNC);
+ String collectionName = message.containsKey(COLLECTION_PROP) ?
+ message.getStr(COLLECTION_PROP) : message.getStr("name");
+ try {
+ try {
+ log.debug("Runner processing {}", head.getId());
+ response = processMessage(message, operation);
+ } finally {
+ timerContext.stop();
+ updateStats();
+ }
+
+ if(asyncId != null) {
+ if (response != null && (response.getResponse().get("failure") != null || response.getResponse().get("exception") != null)) {
+ failureMap.put(asyncId, null);
+ log.debug("Updated failed map for task with zkid:[{}]", head.getId());
+ } else {
+ completedMap.put(asyncId, null);
+ log.debug("Updated completed map for task with zkid:[{}]", head.getId());
+ }
+ } else {
+ head.setBytes(SolrResponse.serializable(response));
+ log.debug("Completed task:[{}]", head.getId());
+ }
+
+ markTaskComplete(head.getId(), asyncId, collectionName);
+ log.debug("Marked task [{}] as completed.", head.getId());
+ printTrackingMaps();
+
+ log.info("Overseer Collection Processor: Message id:" + head.getId() +
+ " complete, response:" + response.getResponse().toString());
+ success = true;
+ } catch (KeeperException e) {
+ SolrException.log(log, "", e);
+ } catch (InterruptedException e) {
+ // Reset task from tracking data structures so that it can be retried.
+ resetTaskWithException(head.getId(), asyncId, collectionName);
+ log.warn("Resetting task {} as the thread was interrupted.", head.getId());
+ Thread.currentThread().interrupt();
+ } finally {
+ if(!success) {
+ // Reset task from tracking data structures so that it can be retried.
+ resetTaskWithException(head.getId(), asyncId, collectionName);
+ }
+ synchronized (waitLock){
+ waitLock.notifyAll();
+ }
+ }
+ }
+
+ private void markTaskComplete(String id, String asyncId, String collectionName)
+ throws KeeperException, InterruptedException {
+ synchronized (completedTasks) {
+ completedTasks.put(id, head);
+ }
+
+ synchronized (runningTasks) {
+ runningTasks.remove(id);
+ }
+
+ if(asyncId != null)
+ runningMap.remove(asyncId);
+
+ if(collectionName != null) {
+ synchronized (collectionWip) {
+ collectionWip.remove(collectionName);
+ }
+ }
+ }
+
+ private void resetTaskWithException(String id, String asyncId, String collectionName) {
+ log.warn("Resetting task: {}, requestid: {}, collectionName: {}", id, asyncId, collectionName);
+ try {
+ if (asyncId != null)
+ runningMap.remove(asyncId);
+
+ synchronized (runningTasks) {
+ runningTasks.remove(id);
+ }
+
+ if (collectionName != null) {
+ synchronized (collectionWip) {
+ collectionWip.remove(collectionName);
+ }
+ }
+ } catch (KeeperException e) {
+ SolrException.log(log, "", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ }
+
+ private void updateStats() {
+ if (isSuccessful()) {
+ stats.success("collection_" + operation);
+ } else {
+ stats.error("collection_" + operation);
+ stats.storeFailureDetails("collection_" + operation, message, response);
+ }
+ }
+
+ private boolean isSuccessful() {
+ if(response == null)
+ return false;
+ return !(response.getResponse().get("failure") != null || response.getResponse().get("exception") != null);
+ }
+ }
+
+ private void printTrackingMaps() {
+ log.debug("RunningTasks: {}", runningTasks.toString());
+ log.debug("CompletedTasks: {}", completedTasks.keySet().toString());
+ log.debug("RunningZKTasks: {}", runningZKTasks.toString());
+ }
+
+
String getId(){
return myId;
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java Tue May 20 19:48:21 2014
@@ -282,16 +282,16 @@ public class CollectionsHandler extends
success.add("state", "completed");
success.add("msg", "found " + requestId + " in completed tasks");
results.add("status", success);
- } else if (coreContainer.getZkController().getOverseerRunningMap().contains(requestId)) {
- SimpleOrderedMap success = new SimpleOrderedMap();
- success.add("state", "running");
- success.add("msg", "found " + requestId + " in submitted tasks");
- results.add("status", success);
} else if (coreContainer.getZkController().getOverseerFailureMap().contains(requestId)) {
SimpleOrderedMap success = new SimpleOrderedMap();
success.add("state", "failed");
success.add("msg", "found " + requestId + " in failed tasks");
results.add("status", success);
+ } else if (coreContainer.getZkController().getOverseerRunningMap().contains(requestId)) {
+ SimpleOrderedMap success = new SimpleOrderedMap();
+ success.add("state", "running");
+ success.add("msg", "found " + requestId + " in submitted tasks");
+ results.add("status", success);
} else {
SimpleOrderedMap failure = new SimpleOrderedMap();
failure.add("state", "notfound");
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/admin/CoreAdminHandler.java Tue May 20 19:48:21 2014
@@ -1346,11 +1346,13 @@ public class CoreAdminHandler extends Re
* Helper method to add a task to a tracking map.
*/
protected void addTask(String map, TaskObject o, boolean limit) {
- if(limit && getMap(map).size() == MAX_TRACKED_REQUESTS) {
- String key = getMap(map).entrySet().iterator().next().getKey();
- getMap(map).remove(key);
+ synchronized (getMap(map)) {
+ if(limit && getMap(map).size() == MAX_TRACKED_REQUESTS) {
+ String key = getMap(map).entrySet().iterator().next().getKey();
+ getMap(map).remove(key);
+ }
+ addTask(map, o);
}
- addTask(map, o);
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java Tue May 20 19:48:21 2014
@@ -405,6 +405,10 @@ public class HttpShardHandler extends Sh
ClientUtils.addSlices(target, collectionName, slices, multiCollection);
}
+ public ShardHandlerFactory getShardHandlerFactory(){
+ return httpShardHandlerFactory;
+ }
+
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/ShardHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/ShardHandler.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/ShardHandler.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/java/org/apache/solr/handler/component/ShardHandler.java Tue May 20 19:48:21 2014
@@ -25,4 +25,5 @@ public abstract class ShardHandler {
public abstract ShardResponse takeCompletedIncludingErrors();
public abstract ShardResponse takeCompletedOrError();
public abstract void cancelAll();
+ public abstract ShardHandlerFactory getShardHandlerFactory();
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerCollectionProcessorTest.java Tue May 20 19:48:21 2014
@@ -31,12 +31,13 @@ import org.apache.solr.common.params.Cor
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.handler.component.ShardHandler;
+import org.apache.solr.handler.component.ShardHandlerFactory;
import org.apache.solr.handler.component.ShardRequest;
import org.apache.solr.handler.component.ShardResponse;
import org.apache.zookeeper.CreateMode;
import org.easymock.Capture;
+import org.easymock.EasyMock;
import org.easymock.IAnswer;
-import org.eclipse.jetty.util.BlockingArrayQueue;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -44,14 +45,17 @@ import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
import static org.easymock.EasyMock.anyBoolean;
import static org.easymock.EasyMock.anyObject;
@@ -74,36 +78,32 @@ public class OverseerCollectionProcessor
private static DistributedMap runningMapMock;
private static DistributedMap completedMapMock;
private static DistributedMap failureMapMock;
+ private static ShardHandlerFactory shardHandlerFactoryMock;
private static ShardHandler shardHandlerMock;
private static ZkStateReader zkStateReaderMock;
private static ClusterState clusterStateMock;
private static SolrZkClient solrZkClientMock;
private final Map zkMap = new HashMap();
private final Set collectionsSet = new HashSet();
+ private SolrResponse lastProcessMessageResult;
+
private OverseerCollectionProcessorToBeTested underTest;
private Thread thread;
- private Queue<QueueEvent> queue = new BlockingArrayQueue<>();
+ private Queue<QueueEvent> queue = new ArrayBlockingQueue<>(10);
private class OverseerCollectionProcessorToBeTested extends
OverseerCollectionProcessor {
- private SolrResponse lastProcessMessageResult;
-
+
public OverseerCollectionProcessorToBeTested(ZkStateReader zkStateReader,
- String myId, ShardHandler shardHandler, String adminPath,
+ String myId, ShardHandlerFactory shardHandlerFactory,
+ String adminPath,
DistributedQueue workQueue, DistributedMap runningMap,
DistributedMap completedMap,
DistributedMap failureMap) {
- super(zkStateReader, myId, shardHandler, adminPath, new Overseer.Stats(), workQueue, runningMap, completedMap, failureMap);
- }
-
- @Override
- protected SolrResponse processMessage(ZkNodeProps message, String operation) {
- lastProcessMessageResult = super.processMessage(message, operation);
- log.info("1 : "+System.currentTimeMillis());
- return lastProcessMessageResult;
+ super(zkStateReader, myId, shardHandlerFactory, adminPath, new Overseer.Stats(), workQueue, runningMap, completedMap, failureMap);
}
@Override
@@ -119,10 +119,12 @@ public class OverseerCollectionProcessor
runningMapMock = createMock(DistributedMap.class);
completedMapMock = createMock(DistributedMap.class);
failureMapMock = createMock(DistributedMap.class);
+ shardHandlerFactoryMock = createMock(ShardHandlerFactory.class);
shardHandlerMock = createMock(ShardHandler.class);
zkStateReaderMock = createMock(ZkStateReader.class);
clusterStateMock = createMock(ClusterState.class);
solrZkClientMock = createMock(SolrZkClient.class);
+
}
@AfterClass
@@ -131,6 +133,7 @@ public class OverseerCollectionProcessor
runningMapMock = null;
completedMapMock = null;
failureMapMock = null;
+ shardHandlerFactoryMock = null;
shardHandlerMock = null;
zkStateReaderMock = null;
clusterStateMock = null;
@@ -145,12 +148,13 @@ public class OverseerCollectionProcessor
reset(runningMapMock);
reset(completedMapMock);
reset(failureMapMock);
+ reset(shardHandlerFactoryMock);
reset(shardHandlerMock);
reset(zkStateReaderMock);
reset(clusterStateMock);
reset(solrZkClientMock);
underTest = new OverseerCollectionProcessorToBeTested(zkStateReaderMock,
- "1234", shardHandlerMock, ADMIN_PATH, workQueueMock, runningMapMock,
+ "1234", shardHandlerFactoryMock, ADMIN_PATH, workQueueMock, runningMapMock,
completedMapMock, failureMapMock);
zkMap.clear();
collectionsSet.clear();
@@ -163,6 +167,44 @@ public class OverseerCollectionProcessor
}
protected Set<String> commonMocks(int liveNodesCount) throws Exception {
+
+ shardHandlerFactoryMock.getShardHandler();
+ expectLastCall().andAnswer(new IAnswer<ShardHandler>() {
+ @Override
+ public ShardHandler answer() throws Throwable {
+ log.info("SHARDHANDLER");
+ return shardHandlerMock;
+ }
+ }).anyTimes();
+ workQueueMock.peekTopN(EasyMock.anyInt(), anyObject(Set.class), EasyMock.anyLong());
+ expectLastCall().andAnswer(new IAnswer<List>() {
+ @Override
+ public List answer() throws Throwable {
+ Object result;
+ int count = 0;
+ while ((result = queue.peek()) == null) {
+ Thread.sleep(1000);
+ count++;
+ if (count > 1) return null;
+ }
+
+ return Arrays.asList(result);
+ }
+ }).anyTimes();
+
+ workQueueMock.getTailId();
+ expectLastCall().andAnswer(new IAnswer<Object>() {
+ @Override
+ public Object answer() throws Throwable {
+ Object result = null;
+ Iterator iter = queue.iterator();
+ while(iter.hasNext()) {
+ result = iter.next();
+ }
+ return result==null ? null : ((QueueEvent)result).getId();
+ }
+ }).anyTimes();
+
workQueueMock.peek(true);
expectLastCall().andAnswer(new IAnswer<Object>() {
@Override
@@ -191,7 +233,7 @@ public class OverseerCollectionProcessor
return queue.poll();
}
}).anyTimes();
-
+
zkStateReaderMock.getClusterState();
expectLastCall().andAnswer(new IAnswer<Object>() {
@Override
@@ -207,8 +249,8 @@ public class OverseerCollectionProcessor
return solrZkClientMock;
}
}).anyTimes();
-
-
+
+
clusterStateMock.getCollections();
expectLastCall().andAnswer(new IAnswer<Object>() {
@Override
@@ -239,7 +281,7 @@ public class OverseerCollectionProcessor
return new HashMap();
}
});
-
+
solrZkClientMock.getZkClientTimeout();
expectLastCall().andAnswer(new IAnswer<Object>() {
@Override
@@ -378,7 +420,12 @@ public class OverseerCollectionProcessor
OverseerCollectionProcessor.MAX_SHARDS_PER_NODE,
maxShardsPerNode.toString());
}
- QueueEvent qe = new QueueEvent("id", ZkStateReader.toJSON(props), null);
+ QueueEvent qe = new QueueEvent("id", ZkStateReader.toJSON(props), null){
+ @Override
+ public void setBytes(byte[] bytes) {
+ lastProcessMessageResult = SolrResponse.deserialize( bytes);
+ }
+ };
queue.add(qe);
}
@@ -542,20 +589,23 @@ public class OverseerCollectionProcessor
replay(solrZkClientMock);
replay(zkStateReaderMock);
replay(clusterStateMock);
+ replay(shardHandlerFactoryMock);
replay(shardHandlerMock);
- log.info("clusterstate " +clusterStateMock.hashCode());
+
+ log.info("clusterstate " + clusterStateMock.hashCode());
startComponentUnderTest();
- issueCreateJob(numberOfSlices, replicationFactor, maxShardsPerNode, (createNodeListOption != CreateNodeListOptions.SEND_NULL)?createNodeList:null, (createNodeListOption != CreateNodeListOptions.DONT_SEND));
+ issueCreateJob(numberOfSlices, replicationFactor, maxShardsPerNode, (createNodeListOption != CreateNodeListOptions.SEND_NULL) ? createNodeList : null, (createNodeListOption != CreateNodeListOptions.DONT_SEND));
waitForEmptyQueue(10000);
if (collectionExceptedToBeCreated) {
- assertNotNull(underTest.lastProcessMessageResult.getResponse().toString(), underTest.lastProcessMessageResult);
+ assertNotNull(lastProcessMessageResult.getResponse().toString(), lastProcessMessageResult);
}
+ verify(shardHandlerFactoryMock);
verify(shardHandlerMock);
-
+
if (collectionExceptedToBeCreated) {
verifySubmitCaptures(submitCaptures, numberOfSlices, replicationFactor,
createNodeList);
Modified: lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerStatusTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerStatusTest.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerStatusTest.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/cloud/OverseerStatusTest.java Tue May 20 19:48:21 2014
@@ -102,22 +102,4 @@ public class OverseerStatusTest extends
waitForThingsToLevelOut(15);
}
-
- private NamedList<Object> invokeCollectionApi(String... args) throws SolrServerException, IOException {
- ModifiableSolrParams params = new ModifiableSolrParams();
- SolrRequest request = new QueryRequest(params);
- for (int i = 0; i < args.length - 1; i+=2) {
- params.add(args[i], args[i+1]);
- }
- request.setPath("/admin/collections");
-
- String baseUrl = ((HttpSolrServer) shardToJetty.get(SHARD1).get(0).client.solrClient)
- .getBaseURL();
- baseUrl = baseUrl.substring(0, baseUrl.length() - "collection1".length());
-
- HttpSolrServer baseServer = new HttpSolrServer(baseUrl);
- baseServer.setConnectionTimeout(15000);
- baseServer.setSoTimeout(60000 * 5);
- return baseServer.request(request);
- }
}
Modified: lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/core/MockShardHandlerFactory.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/core/MockShardHandlerFactory.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/core/MockShardHandlerFactory.java (original)
+++ lucene/dev/branches/branch_4x/solr/core/src/test/org/apache/solr/core/MockShardHandlerFactory.java Tue May 20 19:48:21 2014
@@ -57,6 +57,11 @@ public class MockShardHandlerFactory ext
@Override
public void cancelAll() {}
+
+ @Override
+ public ShardHandlerFactory getShardHandlerFactory() {
+ return MockShardHandlerFactory.this;
+ }
};
}
Modified: lucene/dev/branches/branch_4x/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java (original)
+++ lucene/dev/branches/branch_4x/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java Tue May 20 19:48:21 2014
@@ -88,7 +88,7 @@ public class ZkStateReader {
public static final String LEGACY_CLOUD = "legacyCloud";
public static final String URL_SCHEME = "urlScheme";
-
+
private volatile ClusterState clusterState;
private static final long SOLRCLOUD_UPDATE_DELAY = Long.parseLong(System.getProperty("solrcloud.update.delay", "5000"));
Modified: lucene/dev/branches/branch_4x/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_4x/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java?rev=1596379&r1=1596378&r2=1596379&view=diff
==============================================================================
--- lucene/dev/branches/branch_4x/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java (original)
+++ lucene/dev/branches/branch_4x/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java Tue May 20 19:48:21 2014
@@ -66,6 +66,7 @@ import org.apache.solr.common.cloud.ZkSt
import org.apache.solr.common.params.CollectionParams.CollectionAction;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrCore;
@@ -1711,4 +1712,23 @@ public abstract class AbstractFullDistri
fail("Could not find the new collection - " + exp.code() + " : " + collectionClient.getBaseURL());
}
+ protected NamedList<Object> invokeCollectionApi(String... args) throws SolrServerException, IOException {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ SolrRequest request = new QueryRequest(params);
+ for (int i = 0; i < args.length - 1; i+=2) {
+ params.add(args[i], args[i+1]);
+ }
+ request.setPath("/admin/collections");
+
+ String baseUrl = ((HttpSolrServer) shardToJetty.get(SHARD1).get(0).client.solrClient)
+ .getBaseURL();
+ baseUrl = baseUrl.substring(0, baseUrl.length() - "collection1".length());
+
+ HttpSolrServer baseServer = new HttpSolrServer(baseUrl);
+ baseServer.setConnectionTimeout(15000);
+ baseServer.setSoTimeout(60000 * 5);
+ NamedList r = baseServer.request(request);
+ baseServer.shutdown();
+ return r;
+ }
}