You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2012/08/13 23:22:09 UTC
svn commit: r1372626 - in /lucene/dev/trunk/solr:
core/src/java/org/apache/solr/cloud/
solrj/src/java/org/apache/solr/common/cloud/
solrj/src/java/org/apache/zookeeper/
test-framework/src/java/org/apache/solr/cloud/
Author: markrmiller
Date: Mon Aug 13 21:22:08 2012
New Revision: 1372626
URL: http://svn.apache.org/viewvc?rev=1372626&view=rev
Log:
SOLR-3727: improve solrcloud close/shutdown
Added:
lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ClosableThread.java
- copied, changed from r1372400, lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SafeStopThread.java
Removed:
lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SafeStopThread.java
Modified:
lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java
lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ConnectionManager.java
lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java
lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/Overseer.java Mon Aug 13 21:22:08 2012
@@ -25,6 +25,7 @@ import java.util.Map.Entry;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.ClosableThread;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
@@ -47,7 +48,7 @@ public class Overseer {
private static Logger log = LoggerFactory.getLogger(Overseer.class);
- private class ClusterStateUpdater implements Runnable {
+ private class ClusterStateUpdater implements Runnable, ClosableThread {
private static final String DELETECORE = "deletecore";
private final ZkStateReader reader;
@@ -58,6 +59,7 @@ public class Overseer {
//Internal queue where overseer stores events that have not yet been published into cloudstate
//If Overseer dies while extracting the main queue a new overseer will start from this queue
private final DistributedQueue workQueue;
+ private volatile boolean isClosed;
public ClusterStateUpdater(final ZkStateReader reader, final String myId) {
this.zkClient = reader.getZkClient();
@@ -70,7 +72,7 @@ public class Overseer {
@Override
public void run() {
- if(amILeader() && !Overseer.this.isClosed) {
+ if(!this.isClosed && amILeader()) {
// see if there's something left from the previous Overseer and re
// process all events that were not persisted into cloud state
synchronized (reader.getUpdateLock()) { //XXX this only protects against edits inside single node
@@ -110,7 +112,7 @@ public class Overseer {
}
log.info("Starting to work on the main queue");
- while (amILeader() && !isClosed) {
+ while (!this.isClosed && amILeader()) {
synchronized (reader.getUpdateLock()) {
try {
byte[] head = stateUpdateQueue.peek();
@@ -399,12 +401,48 @@ public class Overseer {
ClusterState newState = new ClusterState(clusterState.getLiveNodes(), newStates);
return newState;
}
+
+ @Override
+ public void close() {
+ this.isClosed = true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.isClosed;
+ }
}
- private Thread ccThread;
+ class OverseerThread extends Thread implements ClosableThread {
+
+ private volatile boolean isClosed;
+
+ public OverseerThread(ThreadGroup tg,
+ ClusterStateUpdater clusterStateUpdater) {
+ super(tg, clusterStateUpdater);
+ }
+
+ public OverseerThread(ThreadGroup ccTg,
+ OverseerCollectionProcessor overseerCollectionProcessor, String string) {
+ super(ccTg, overseerCollectionProcessor, string);
+ }
+
+ @Override
+ public void close() {
+ this.isClosed = true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.isClosed;
+ }
+
+ }
+
+ private OverseerThread ccThread;
- private Thread updaterThread;
+ private OverseerThread updaterThread;
private volatile boolean isClosed;
@@ -425,11 +463,11 @@ public class Overseer {
createOverseerNode(reader.getZkClient());
//launch cluster state updater thread
ThreadGroup tg = new ThreadGroup("Overseer state updater.");
- updaterThread = new Thread(tg, new ClusterStateUpdater(reader, id));
+ updaterThread = new OverseerThread(tg, new ClusterStateUpdater(reader, id));
updaterThread.setDaemon(true);
ThreadGroup ccTg = new ThreadGroup("Overseer collection creation process.");
- ccThread = new Thread(ccTg, new OverseerCollectionProcessor(reader, id, shardHandler, adminPath),
+ ccThread = new OverseerThread(ccTg, new OverseerCollectionProcessor(reader, id, shardHandler, adminPath),
"Overseer-" + id);
ccThread.setDaemon(true);
@@ -439,6 +477,12 @@ public class Overseer {
public void close() {
isClosed = true;
+ if (updaterThread != null) {
+ updaterThread.close();
+ }
+ if (ccThread != null) {
+ ccThread.close();
+ }
}
/**
Modified: lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java (original)
+++ lucene/dev/trunk/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java Mon Aug 13 21:22:08 2012
@@ -31,7 +31,7 @@ import org.apache.solr.client.solrj.requ
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.SafeStopThread;
+import org.apache.solr.common.cloud.ClosableThread;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -56,7 +56,7 @@ import org.apache.zookeeper.KeeperExcept
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class RecoveryStrategy extends Thread implements SafeStopThread {
+public class RecoveryStrategy extends Thread implements ClosableThread {
private static final int MAX_RETRIES = 500;
private static final int INTERRUPTED = MAX_RETRIES + 1;
private static final int STARTING_RECOVERY_DELAY = 1000;
Copied: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ClosableThread.java (from r1372400, lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SafeStopThread.java)
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ClosableThread.java?p2=lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ClosableThread.java&p1=lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SafeStopThread.java&r1=1372400&r2=1372626&rev=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SafeStopThread.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ClosableThread.java Mon Aug 13 21:22:08 2012
@@ -17,7 +17,7 @@ package org.apache.solr.common.cloud;
* limitations under the License.
*/
-public interface SafeStopThread {
- public void stop();
+public interface ClosableThread {
+ public void close();
public boolean isClosed();
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ConnectionManager.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ConnectionManager.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ConnectionManager.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ConnectionManager.java Mon Aug 13 21:22:08 2012
@@ -71,7 +71,9 @@ class ConnectionManager implements Watch
+ " path:" + event.getPath() + " type:" + event.getType());
}
- checkClosed();
+ if (isClosed) {
+ return;
+ }
state = event.getState();
if (state == KeeperState.SyncConnected) {
@@ -85,18 +87,21 @@ class ConnectionManager implements Watch
connectionStrategy.reconnect(zkServerAddress, zkClientTimeout, this,
new ZkClientConnectionStrategy.ZkUpdate() {
@Override
- public void update(SolrZooKeeper keeper) throws TimeoutException {
+ public void update(SolrZooKeeper keeper) {
+ // if keeper does not replace oldKeeper we must be sure to close it
synchronized (connectionStrategy) {
- checkClosed();
try {
waitForConnected(SolrZkClient.DEFAULT_CLIENT_CONNECT_TIMEOUT);
- checkClosed();
client.updateKeeper(keeper);
} catch (InterruptedException e) {
+ closeKeeper(keeper);
// we must have been asked to stop
throw new RuntimeException("Giving up on connecting - we were interrupted");
+ } catch(Throwable t) {
+ closeKeeper(keeper);
+ throw new RuntimeException(t);
}
- checkClosed();
+
if (onReconnect != null) {
onReconnect.command();
}
@@ -140,20 +145,15 @@ class ConnectionManager implements Watch
long left = waitForConnection;
while (!connected && left > 0) {
wait(left);
- checkClosed();
+ if (isClosed) {
+ break;
+ }
left = expire - System.currentTimeMillis();
}
if (!connected) {
throw new TimeoutException("Could not connect to ZooKeeper " + zkServerAddress + " within " + waitForConnection + " ms");
}
}
-
- private synchronized void checkClosed() {
- if (isClosed) {
- log.info("Not acting because I am closed");
- return;
- }
- }
public synchronized void waitForDisconnected(long timeout)
throws InterruptedException, TimeoutException {
@@ -167,4 +167,16 @@ class ConnectionManager implements Watch
throw new TimeoutException("Did not disconnect");
}
}
+
+ private void closeKeeper(SolrZooKeeper keeper) {
+ try {
+ keeper.close();
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.error("", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
+ "", e);
+ }
+ }
}
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java Mon Aug 13 21:22:08 2012
@@ -137,6 +137,18 @@ public class SolrZkClient {
"", e);
}
}
+ if (isClosed) {
+ // we may have been closed
+ try {
+ SolrZkClient.this.keeper.close();
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.error("", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
+ "", e);
+ }
+ }
}
});
connManager.waitForConnected(clientConnectTimeout);
@@ -673,11 +685,13 @@ public class SolrZkClient {
* @throws InterruptedException
*/
void updateKeeper(SolrZooKeeper keeper) throws InterruptedException {
+ if (isClosed) throw new RuntimeException("client is closed");
SolrZooKeeper oldKeeper = this.keeper;
this.keeper = keeper;
if (oldKeeper != null) {
oldKeeper.close();
}
+ if (isClosed) this.keeper.close();
}
public SolrZooKeeper getSolrZooKeeper() {
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/common/cloud/ZkCmdExecutor.java Mon Aug 13 21:22:08 2012
@@ -73,9 +73,9 @@ public class ZkCmdExecutor {
Thread.currentThread().interrupt();
throw new InterruptedException();
}
- if (Thread.currentThread() instanceof SafeStopThread) {
- if (((SafeStopThread) Thread.currentThread()).isClosed()) {
- throw new RuntimeException("Interrupted");
+ if (Thread.currentThread() instanceof ClosableThread) {
+ if (((ClosableThread) Thread.currentThread()).isClosed()) {
+ throw exception;
}
}
retryDelay(i);
Modified: lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java (original)
+++ lucene/dev/trunk/solr/solrj/src/java/org/apache/zookeeper/SolrZooKeeper.java Mon Aug 13 21:22:08 2012
@@ -19,16 +19,23 @@ package org.apache.zookeeper;
import java.io.IOException;
import java.nio.channels.SocketChannel;
+import java.util.Iterator;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
// we use this class to expose nasty stuff for tests
public class SolrZooKeeper extends ZooKeeper {
List<Thread> spawnedThreads = new CopyOnWriteArrayList<Thread>();
+
+ // for test debug
+ //static Map<SolrZooKeeper,Exception> clients = new ConcurrentHashMap<SolrZooKeeper,Exception>();
public SolrZooKeeper(String connectString, int sessionTimeout,
Watcher watcher) throws IOException {
super(connectString, sessionTimeout, watcher);
+ //clients.put(this, new RuntimeException());
}
public ClientCnxn getConnection() {
@@ -64,9 +71,20 @@ public class SolrZooKeeper extends ZooKe
@Override
public synchronized void close() throws InterruptedException {
+ //clients.remove(this);
for (Thread t : spawnedThreads) {
t.interrupt();
}
super.close();
}
+
+// public static void assertCloses() {
+// if (clients.size() > 0) {
+// Iterator<Exception> stacktraces = clients.values().iterator();
+// Exception cause = null;
+// cause = stacktraces.next();
+// throw new RuntimeException("Found a bad one!", cause);
+// }
+// }
+
}
Modified: lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java?rev=1372626&r1=1372625&r2=1372626&view=diff
==============================================================================
--- lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java (original)
+++ lucene/dev/trunk/solr/test-framework/src/java/org/apache/solr/cloud/AbstractFullDistribZkTestBase.java Mon Aug 13 21:22:08 2012
@@ -31,9 +31,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.http.params.CoreConnectionPNames;
-import org.apache.lucene.util.LuceneTestCase.BadApple;
import org.apache.lucene.util.LuceneTestCase.Slow;
-import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.client.solrj.SolrServerException;
@@ -42,7 +40,6 @@ import org.apache.solr.client.solrj.impl
import org.apache.solr.client.solrj.impl.HttpSolrServer;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
-import org.apache.solr.cloud.ChaosMonkey;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.SolrException;
@@ -65,8 +62,6 @@ import org.slf4j.LoggerFactory;
* what we test now - the default update chain
*/
@Slow
-@BadApple
-@AwaitsFix(bugUrl = "SOLR-3727 (leak threads)")
public abstract class AbstractFullDistribZkTestBase extends AbstractDistribZkTestBase {
static Logger log = LoggerFactory.getLogger(AbstractFullDistribZkTestBase.class);