You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by si...@apache.org on 2012/09/21 19:22:27 UTC
svn commit: r1388574 [35/45] - in /lucene/dev/branches/LUCENE-2878: ./
dev-tools/ dev-tools/eclipse/ dev-tools/eclipse/dot.settings/
dev-tools/idea/ dev-tools/idea/.idea/ dev-tools/idea/.idea/libraries/
dev-tools/idea/lucene/ dev-tools/idea/lucene/anal...
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java Fri Sep 21 17:21:34 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;
@@ -65,8 +65,14 @@ public class RecoveryStrategy extends Th
private static Logger log = LoggerFactory.getLogger(RecoveryStrategy.class);
+ public static interface RecoveryListener {
+ public void recovered();
+ public void failed();
+ }
+
private volatile boolean close = false;
+ private RecoveryListener recoveryListener;
private ZkController zkController;
private String baseUrl;
private String coreZkNodeName;
@@ -76,9 +82,10 @@ public class RecoveryStrategy extends Th
private boolean recoveringAfterStartup;
private CoreContainer cc;
- public RecoveryStrategy(CoreContainer cc, String name) {
+ public RecoveryStrategy(CoreContainer cc, String name, RecoveryListener recoveryListener) {
this.cc = cc;
this.coreName = name;
+ this.recoveryListener = recoveryListener;
setName("RecoveryThread");
zkController = cc.getZkController();
zkStateReader = zkController.getZkStateReader();
@@ -93,7 +100,7 @@ public class RecoveryStrategy extends Th
// make sure any threads stop retrying
public void close() {
close = true;
- log.warn("Stopping recovery for core " + coreName + " zkNodeName=" + coreZkNodeName);
+ log.warn("Stopping recovery for zkNodeName=" + coreZkNodeName + "core=" + coreName );
}
@@ -105,13 +112,14 @@ public class RecoveryStrategy extends Th
zkController.publish(cd, ZkStateReader.RECOVERY_FAILED);
} finally {
close();
+ recoveryListener.failed();
}
}
private void replicate(String nodeName, SolrCore core, ZkNodeProps leaderprops, String baseUrl)
throws SolrServerException, IOException {
- String leaderBaseUrl = leaderprops.get(ZkStateReader.BASE_URL_PROP);
+ String leaderBaseUrl = leaderprops.getStr(ZkStateReader.BASE_URL_PROP);
ZkCoreNodeProps leaderCNodeProps = new ZkCoreNodeProps(leaderprops);
String leaderUrl = leaderCNodeProps.getCoreUrl();
@@ -120,6 +128,7 @@ public class RecoveryStrategy extends Th
// if we are the leader, either we are trying to recover faster
// then our ephemeral timed out or we are the only node
if (!leaderBaseUrl.equals(baseUrl)) {
+
// send commit
commitOnLeader(leaderUrl);
@@ -139,7 +148,7 @@ public class RecoveryStrategy extends Th
solrParams.set(ReplicationHandler.MASTER_URL, leaderUrl);
if (isClosed()) retries = INTERRUPTED;
- boolean success = replicationHandler.doFetch(solrParams, true); // TODO: look into making sure force=true does not download files we already have
+ boolean success = replicationHandler.doFetch(solrParams, true); // TODO: look into making force=true not download files we already have?
if (!success) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Replication for recovery failed.");
@@ -185,7 +194,7 @@ public class RecoveryStrategy extends Th
prepCmd.setCoreNodeName(coreZkNodeName);
prepCmd.setState(ZkStateReader.RECOVERING);
prepCmd.setCheckLive(true);
- prepCmd.setPauseFor(6000);
+ prepCmd.setOnlyIfLeader(true);
server.request(prepCmd);
server.shutdown();
@@ -209,15 +218,15 @@ public class RecoveryStrategy extends Th
try {
doRecovery(core);
- } catch (KeeperException e) {
- log.error("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- } catch (InterruptedException e) {
+ } catch (InterruptedException e) {
Thread.currentThread().interrupt();
SolrException.log(log, "", e);
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "",
e);
+ } catch (Throwable t) {
+ log.error("", t);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
+ "", t);
}
} finally {
if (core != null) core.close();
@@ -239,6 +248,7 @@ public class RecoveryStrategy extends Th
return;
}
+ boolean firstTime = true;
List<Long> recentVersions;
UpdateLog.RecentUpdates recentUpdates = null;
@@ -256,57 +266,78 @@ public class RecoveryStrategy extends Th
List<Long> startingVersions = ulog.getStartingVersions();
-
if (startingVersions != null && recoveringAfterStartup) {
- int oldIdx = 0; // index of the start of the old list in the current list
- long firstStartingVersion = startingVersions.size() > 0 ? startingVersions.get(0) : 0;
-
- for (; oldIdx<recentVersions.size(); oldIdx++) {
- if (recentVersions.get(oldIdx) == firstStartingVersion) break;
- }
-
- if (oldIdx > 0) {
- log.info("####### Found new versions added after startup: num=" + oldIdx);
- log.info("###### currentVersions=" + recentVersions);
+ try {
+ int oldIdx = 0; // index of the start of the old list in the current
+ // list
+ long firstStartingVersion = startingVersions.size() > 0 ? startingVersions
+ .get(0) : 0;
+
+ for (; oldIdx < recentVersions.size(); oldIdx++) {
+ if (recentVersions.get(oldIdx) == firstStartingVersion) break;
+ }
+
+ if (oldIdx > 0) {
+ log.info("####### Found new versions added after startup: num="
+ + oldIdx);
+ log.info("###### currentVersions=" + recentVersions);
+ }
+
+ log.info("###### startupVersions=" + startingVersions);
+ } catch (Throwable t) {
+ SolrException.log(log, "Error getting recent versions. core=" + coreName, t);
+ recentVersions = new ArrayList<Long>(0);
}
-
- log.info("###### startupVersions=" + startingVersions);
}
-
- boolean firstTime = true;
-
if (recoveringAfterStartup) {
// if we're recovering after startup (i.e. we have been down), then we need to know what the last versions were
// when we went down. We may have received updates since then.
recentVersions = startingVersions;
-
- if ((ulog.getStartingOperation() & UpdateLog.FLAG_GAP) != 0) {
- // last operation at the time of startup had the GAP flag set...
- // this means we were previously doing a full index replication
- // that probably didn't complete and buffering updates in the meantime.
- log.info("Looks like a previous replication recovery did not complete - skipping peer sync. core=" + coreName);
- firstTime = false; // skip peersync
+ try {
+ if ((ulog.getStartingOperation() & UpdateLog.FLAG_GAP) != 0) {
+ // last operation at the time of startup had the GAP flag set...
+ // this means we were previously doing a full index replication
+ // that probably didn't complete and buffering updates in the
+ // meantime.
+ log.info("Looks like a previous replication recovery did not complete - skipping peer sync. core="
+ + coreName);
+ firstTime = false; // skip peersync
+ }
+ } catch (Throwable t) {
+ SolrException.log(log, "Error trying to get ulog starting operation. core="
+ + coreName, t);
+ firstTime = false; // skip peersync
}
}
- while (!successfulRecovery && !isClosed() && !isInterrupted()) { // don't use interruption or it will close channels though
+ while (!successfulRecovery && !isInterrupted()) { // don't use interruption or it will close channels though
try {
- // first thing we just try to sync
- zkController.publish(core.getCoreDescriptor(), ZkStateReader.RECOVERING);
-
CloudDescriptor cloudDesc = core.getCoreDescriptor()
.getCloudDescriptor();
ZkNodeProps leaderprops = zkStateReader.getLeaderProps(
cloudDesc.getCollectionName(), cloudDesc.getShardId());
-
- String leaderBaseUrl = leaderprops.get(ZkStateReader.BASE_URL_PROP);
- String leaderCoreName = leaderprops.get(ZkStateReader.CORE_NAME_PROP);
+
+ String leaderBaseUrl = leaderprops.getStr(ZkStateReader.BASE_URL_PROP);
+ String leaderCoreName = leaderprops.getStr(ZkStateReader.CORE_NAME_PROP);
String leaderUrl = ZkCoreNodeProps.getCoreUrl(leaderBaseUrl, leaderCoreName);
- sendPrepRecoveryCmd(leaderBaseUrl, leaderCoreName);
+ String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
+ boolean isLeader = leaderUrl.equals(ourUrl);
+ if (isLeader && !cloudDesc.isLeader) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Cloud state still says we are leader.");
+ }
+ if (cloudDesc.isLeader) {
+ // we are now the leader - no one else must have been suitable
+ log.warn("We have not yet recovered - but we are now the leader! core=" + coreName);
+ log.info("Finished recovery process. core=" + coreName);
+ zkController.publish(core.getCoreDescriptor(), ZkStateReader.ACTIVE);
+ return;
+ }
+
+ zkController.publish(core.getCoreDescriptor(), ZkStateReader.RECOVERING);
// first thing we just try to sync
if (firstTime) {
@@ -321,11 +352,9 @@ public class RecoveryStrategy extends Th
if (syncSuccess) {
SolrQueryRequest req = new LocalSolrQueryRequest(core,
new ModifiableSolrParams());
+ // force open a new searcher
core.getUpdateHandler().commit(new CommitUpdateCommand(req, false));
log.info("PeerSync Recovery was successful - registering as Active. core=" + coreName);
- // System.out
- // .println("Sync Recovery was successful - registering as Active "
- // + zkController.getNodeName());
// solrcloud_debug
// try {
@@ -353,12 +382,24 @@ public class RecoveryStrategy extends Th
log.info("PeerSync Recovery was not successful - trying replication. core=" + coreName);
}
- //System.out.println("Sync Recovery was not successful - trying replication");
+
log.info("Starting Replication Recovery. core=" + coreName);
+
+ sendPrepRecoveryCmd(leaderBaseUrl, leaderCoreName);
+
+ // we wait a bit so that any updates on the leader
+ // that started before they saw recovering state
+ // are sure to have finished
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
log.info("Begin buffering updates. core=" + coreName);
ulog.bufferUpdates();
replayed = false;
-
+
try {
replicate(zkController.getNodeName(), core,
@@ -372,12 +413,13 @@ public class RecoveryStrategy extends Th
zkController.publish(core.getCoreDescriptor(), ZkStateReader.ACTIVE);
close = true;
successfulRecovery = true;
+ recoveryListener.recovered();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("Recovery was interrupted", e);
retries = INTERRUPTED;
} catch (Throwable t) {
- log.error("Error while trying to recover", t);
+ SolrException.log(log, "Error while trying to recover", t);
} finally {
if (!replayed) {
try {
@@ -390,7 +432,7 @@ public class RecoveryStrategy extends Th
}
} catch (Throwable t) {
- log.error("Error while trying to recover. core=" + coreName, t);
+ SolrException.log(log, "Error while trying to recover. core=" + coreName, t);
}
if (!successfulRecovery) {
@@ -400,20 +442,39 @@ public class RecoveryStrategy extends Th
try {
log.error("Recovery failed - trying again... core=" + coreName);
+
+ if (isClosed()) {
+ retries = INTERRUPTED;
+ }
+
retries++;
if (retries >= MAX_RETRIES) {
if (retries == INTERRUPTED) {
-
+ SolrException.log(log, "Recovery failed - interrupted. core="
+ + coreName);
+ try {
+ recoveryFailed(core, zkController, baseUrl, coreZkNodeName,
+ core.getCoreDescriptor());
+ } catch (Throwable t) {
+ SolrException.log(log,
+ "Could not publish that recovery failed", t);
+ }
} else {
- log.error("Recovery failed - max retries exceeded. core=" + coreName);
- recoveryFailed(core, zkController, baseUrl, coreZkNodeName,
- core.getCoreDescriptor());
+ SolrException.log(log,
+ "Recovery failed - max retries exceeded. core=" + coreName);
+ try {
+ recoveryFailed(core, zkController, baseUrl, coreZkNodeName,
+ core.getCoreDescriptor());
+ } catch (Throwable t) {
+ SolrException.log(log,
+ "Could not publish that recovery failed", t);
+ }
}
break;
}
- } catch (Exception e) {
- log.error("core=" + coreName, e);
+ } catch (Throwable e) {
+ SolrException.log(log, "core=" + coreName, e);
}
try {
@@ -444,7 +505,11 @@ public class RecoveryStrategy extends Th
} else {
log.info("Replaying buffered documents. core=" + coreName);
// wait for replay
- future.get();
+ RecoveryInfo report = future.get();
+ if (report.failed) {
+ SolrException.log(log, "Replay failed");
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Replay failed");
+ }
}
// solrcloud_debug
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/SyncStrategy.java Fri Sep 21 17:21:34 2012
@@ -20,7 +20,6 @@ package org.apache.solr.cloud;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import org.apache.http.client.HttpClient;
import org.apache.solr.client.solrj.SolrServerException;
@@ -28,8 +27,6 @@ import org.apache.solr.client.solrj.impl
import org.apache.solr.client.solrj.impl.HttpSolrServer;
import org.apache.solr.client.solrj.request.CoreAdminRequest.RequestRecovery;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -48,6 +45,8 @@ import org.slf4j.LoggerFactory;
public class SyncStrategy {
protected final Logger log = LoggerFactory.getLogger(getClass());
+ private final boolean SKIP_AUTO_RECOVERY = Boolean.getBoolean("solrcloud.skip.autorecovery");
+
private final ShardHandler shardHandler;
private final static HttpClient client;
@@ -72,6 +71,9 @@ public class SyncStrategy {
public boolean sync(ZkController zkController, SolrCore core,
ZkNodeProps leaderProps) {
+ if (SKIP_AUTO_RECOVERY) {
+ return true;
+ }
log.info("Sync replicas to " + ZkCoreNodeProps.getCoreUrl(leaderProps));
// TODO: look at our state usage of sync
// zkController.publish(core, ZkStateReader.SYNC);
@@ -93,6 +95,21 @@ public class SyncStrategy {
String collection = cloudDesc.getCollectionName();
String shardId = cloudDesc.getShardId();
+ // if no one that is up is active, we are willing to wait...
+ // we don't want a recovering node to become leader and then
+ // a better candidate pops up a second later.
+// int tries = 20;
+// while (!areAnyReplicasActive(zkController, collection, shardId)) {
+// if (tries-- == 0) {
+// break;
+// }
+// try {
+// Thread.sleep(500);
+// } catch (InterruptedException e) {
+// Thread.currentThread().interrupt();
+// }
+// }
+
// first sync ourselves - we are the potential leader after all
try {
success = syncWithReplicas(zkController, core, leaderProps, collection,
@@ -101,17 +118,7 @@ public class SyncStrategy {
SolrException.log(log, "Sync Failed", e);
}
try {
- // if !success but no one else is in active mode,
- // we are the leader anyway
- // TODO: should we also be leader if there is only one other active?
- // if we couldn't sync with it, it shouldn't be able to sync with us
- if (!success
- && !areAnyOtherReplicasActive(zkController, leaderProps, collection,
- shardId)) {
- log.info("Sync was not a success but no on else i active! I am the leader");
- success = true;
- }
-
+
if (success) {
log.info("Sync Success - now sync replicas to me");
@@ -119,7 +126,6 @@ public class SyncStrategy {
} else {
SolrException.log(log, "Sync Failed");
-
// lets see who seems ahead...
}
@@ -130,39 +136,12 @@ public class SyncStrategy {
return success;
}
- private boolean areAnyOtherReplicasActive(ZkController zkController,
- ZkNodeProps leaderProps, String collection, String shardId) {
- ClusterState clusterState = zkController.getZkStateReader().getClusterState();
- Map<String,Slice> slices = clusterState.getSlices(collection);
- Slice slice = slices.get(shardId);
- Map<String,ZkNodeProps> shards = slice.getShards();
- for (Map.Entry<String,ZkNodeProps> shard : shards.entrySet()) {
- String state = shard.getValue().get(ZkStateReader.STATE_PROP);
-// System.out.println("state:"
-// + state
-// + shard.getValue().get(ZkStateReader.NODE_NAME_PROP)
-// + " live: "
-// + clusterState.liveNodesContain(shard.getValue().get(
-// ZkStateReader.NODE_NAME_PROP)));
- if ((state.equals(ZkStateReader.ACTIVE))
- && clusterState.liveNodesContain(shard.getValue().get(
- ZkStateReader.NODE_NAME_PROP))
- && !new ZkCoreNodeProps(shard.getValue()).getCoreUrl().equals(
- new ZkCoreNodeProps(leaderProps).getCoreUrl())) {
- return true;
- }
- }
-
- return false;
- }
-
private boolean syncWithReplicas(ZkController zkController, SolrCore core,
ZkNodeProps props, String collection, String shardId) {
List<ZkCoreNodeProps> nodes = zkController.getZkStateReader()
.getReplicaProps(collection, shardId,
- props.get(ZkStateReader.NODE_NAME_PROP),
- props.get(ZkStateReader.CORE_NAME_PROP), ZkStateReader.ACTIVE); // TODO:
- // TODO should there be a state filter?
+ props.getStr(ZkStateReader.NODE_NAME_PROP),
+ props.getStr(ZkStateReader.CORE_NAME_PROP));
if (nodes == null) {
// I have no replicas
@@ -171,14 +150,13 @@ public class SyncStrategy {
List<String> syncWith = new ArrayList<String>();
for (ZkCoreNodeProps node : nodes) {
- // if we see a leader, must be stale state, and this is the guy that went down
- if (!node.getNodeProps().keySet().contains(ZkStateReader.LEADER_PROP)) {
- syncWith.add(node.getCoreUrl());
- }
+ syncWith.add(node.getCoreUrl());
}
-
- PeerSync peerSync = new PeerSync(core, syncWith, core.getUpdateHandler().getUpdateLog().numRecordsToKeep);
+ // if we can't reach a replica for sync, we still consider the overall sync a success
+ // TODO: as an assurance, we should still try and tell the sync nodes that we couldn't reach
+ // to recover once more?
+ PeerSync peerSync = new PeerSync(core, syncWith, core.getUpdateHandler().getUpdateLog().numRecordsToKeep, true);
return peerSync.sync();
}
@@ -190,8 +168,8 @@ public class SyncStrategy {
List<ZkCoreNodeProps> nodes = zkController
.getZkStateReader()
.getReplicaProps(collection, shardId,
- leaderProps.get(ZkStateReader.NODE_NAME_PROP),
- leaderProps.get(ZkStateReader.CORE_NAME_PROP), ZkStateReader.ACTIVE);
+ leaderProps.getStr(ZkStateReader.NODE_NAME_PROP),
+ leaderProps.getStr(ZkStateReader.CORE_NAME_PROP));
if (nodes == null) {
log.info(ZkCoreNodeProps.getCoreUrl(leaderProps) + " has no replicas");
return;
@@ -222,16 +200,16 @@ public class SyncStrategy {
try {
log.info(ZkCoreNodeProps.getCoreUrl(leaderProps) + ": Sync failed - asking replica (" + srsp.getShardAddress() + ") to recover.");
- requestRecovery(((ShardCoreRequest)srsp.getShardRequest()).baseUrl, ((ShardCoreRequest)srsp.getShardRequest()).coreName);
+ requestRecovery(leaderProps, ((ShardCoreRequest)srsp.getShardRequest()).baseUrl, ((ShardCoreRequest)srsp.getShardRequest()).coreName);
- } catch (Exception e) {
- SolrException.log(log, ZkCoreNodeProps.getCoreUrl(leaderProps) + ": Could not tell a replica to recover", e);
+ } catch (Throwable t) {
+ SolrException.log(log, ZkCoreNodeProps.getCoreUrl(leaderProps) + ": Could not tell a replica to recover", t);
}
} else {
log.info(ZkCoreNodeProps.getCoreUrl(leaderProps) + ": " + " sync completed with " + srsp.getShardAddress());
}
+
}
-
}
@@ -269,16 +247,29 @@ public class SyncStrategy {
shardHandler.submit(sreq, replica, sreq.params);
}
- private void requestRecovery(String baseUrl, String coreName) throws SolrServerException, IOException {
+ private void requestRecovery(final ZkNodeProps leaderProps, final String baseUrl, final String coreName) throws SolrServerException, IOException {
// TODO: do this in background threads
- RequestRecovery recoverRequestCmd = new RequestRecovery();
- recoverRequestCmd.setAction(CoreAdminAction.REQUESTRECOVERY);
- recoverRequestCmd.setCoreName(coreName);
-
- HttpSolrServer server = new HttpSolrServer(baseUrl);
- server.setConnectionTimeout(45000);
- server.setSoTimeout(45000);
- server.request(recoverRequestCmd);
+ Thread thread = new Thread() {
+ {
+ setDaemon(true);
+ }
+ @Override
+ public void run() {
+ RequestRecovery recoverRequestCmd = new RequestRecovery();
+ recoverRequestCmd.setAction(CoreAdminAction.REQUESTRECOVERY);
+ recoverRequestCmd.setCoreName(coreName);
+
+ HttpSolrServer server = new HttpSolrServer(baseUrl);
+ server.setConnectionTimeout(45000);
+ server.setSoTimeout(45000);
+ try {
+ server.request(recoverRequestCmd);
+ } catch (Throwable t) {
+ SolrException.log(log, ZkCoreNodeProps.getCoreUrl(leaderProps) + ": Could not tell a replica to recover", t);
+ }
+ }
+ };
+ thread.run();
}
public static ModifiableSolrParams params(String... params) {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/ZkController.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/ZkController.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/ZkController.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/cloud/ZkController.java Fri Sep 21 17:21:34 2012
@@ -27,6 +27,9 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Future;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -47,18 +50,20 @@ import org.apache.solr.common.cloud.ZkNo
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.ZooKeeperException;
import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+
import org.apache.solr.core.Config;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.CoreDescriptor;
import org.apache.solr.core.SolrCore;
-import org.apache.solr.handler.component.HttpShardHandlerFactory;
-
import org.apache.solr.handler.component.ShardHandler;
import org.apache.solr.update.UpdateLog;
import org.apache.solr.util.DOMUtil;
+import org.apache.solr.util.DefaultSolrThreadFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -96,6 +101,11 @@ public final class ZkController {
public final static String COLLECTION_PARAM_PREFIX="collection.";
public final static String CONFIGNAME_PROP="configName";
+ private ThreadPoolExecutor cmdDistribExecutor = new ThreadPoolExecutor(
+ 0, Integer.MAX_VALUE, 5, TimeUnit.SECONDS,
+ new SynchronousQueue<Runnable>(), new DefaultSolrThreadFactory(
+ "cmdDistribExecutor"));
+
private final Map<String, ElectionContext> electionContexts = Collections.synchronizedMap(new HashMap<String, ElectionContext>());
private SolrZkClient zkClient;
@@ -123,8 +133,13 @@ public final class ZkController {
protected volatile Overseer overseer;
+ private String leaderVoteWait;
+
+ private int clientTimeout;
+
+
/**
- * @param cc if null, recovery will not be enabled
+ * @param cc
* @param zkServerAddress
* @param zkClientTimeout
* @param zkClientConnectTimeout
@@ -139,6 +154,28 @@ public final class ZkController {
public ZkController(final CoreContainer cc, String zkServerAddress, int zkClientTimeout, int zkClientConnectTimeout, String localHost, String locaHostPort,
String localHostContext, final CurrentCoreDescriptorProvider registerOnReconnect) throws InterruptedException,
TimeoutException, IOException {
+ this(cc, zkServerAddress, zkClientTimeout, zkClientConnectTimeout, localHost, locaHostPort, localHostContext, null, registerOnReconnect);
+ }
+
+
+ /**
+ * @param cc
+ * @param zkServerAddress
+ * @param zkClientTimeout
+ * @param zkClientConnectTimeout
+ * @param localHost
+ * @param locaHostPort
+ * @param localHostContext
+ * @param leaderVoteWait
+ * @param registerOnReconnect
+ * @throws InterruptedException
+ * @throws TimeoutException
+ * @throws IOException
+ */
+ public ZkController(final CoreContainer cc, String zkServerAddress, int zkClientTimeout, int zkClientConnectTimeout, String localHost, String locaHostPort,
+ String localHostContext, String leaderVoteWait, final CurrentCoreDescriptorProvider registerOnReconnect) throws InterruptedException,
+ TimeoutException, IOException {
+ if (cc == null) throw new IllegalArgumentException("CoreContainer cannot be null.");
this.cc = cc;
if (localHostContext.contains("/")) {
throw new IllegalArgumentException("localHostContext ("
@@ -152,34 +189,40 @@ public final class ZkController {
this.hostName = getHostNameFromAddress(this.localHost);
this.nodeName = this.hostName + ':' + this.localHostPort + '_' + this.localHostContext;
this.baseURL = this.localHost + ":" + this.localHostPort + "/" + this.localHostContext;
-
+ this.leaderVoteWait = leaderVoteWait;
+ this.clientTimeout = zkClientTimeout;
zkClient = new SolrZkClient(zkServerAddress, zkClientTimeout, zkClientConnectTimeout,
// on reconnect, reload cloud info
new OnReconnect() {
public void command() {
try {
+ markAllAsNotLeader(registerOnReconnect);
+
+ // this is troublesome - we dont want to kill anything the old leader accepted
+ // though I guess sync will likely get those updates back? But only if
+ // he is involved in the sync, and he certainly may not be
+ // ExecutorUtil.shutdownAndAwaitTermination(cc.getCmdDistribExecutor());
// we need to create all of our lost watches
// seems we dont need to do this again...
//Overseer.createClientNodes(zkClient, getNodeName());
ShardHandler shardHandler;
String adminPath;
- if (cc == null) {
- shardHandler = new HttpShardHandlerFactory().getShardHandler();
- adminPath = "/admin/cores";
- } else {
- shardHandler = cc.getShardHandlerFactory().getShardHandler();
- adminPath = cc.getAdminPath();
- }
+ shardHandler = cc.getShardHandlerFactory().getShardHandler();
+ adminPath = cc.getAdminPath();
+
+ cc.cancelCoreRecoveries();
+
+ registerAllCoresAsDown(registerOnReconnect, false);
+
ZkController.this.overseer = new Overseer(shardHandler, adminPath, zkStateReader);
ElectionContext context = new OverseerElectionContext(zkClient, overseer, getNodeName());
overseerElector.joinElection(context);
zkStateReader.createClusterStateWatchersAndUpdate();
-
- registerAllCoresAsDown(registerOnReconnect);
-
+ // cc.newCmdDistribExecutor();
+
// we have to register as live first to pick up docs in the buffer
createEphemeralLiveNode();
@@ -190,7 +233,11 @@ public final class ZkController {
// TODO: we need to think carefully about what happens when it was
// a leader that was expired - as well as what to do about leaders/overseers
// with connection loss
- register(descriptor.getName(), descriptor, true);
+ try {
+ register(descriptor.getName(), descriptor, true, true);
+ } catch (Throwable t) {
+ SolrException.log(log, "Error registering SolrCore", t);
+ }
}
}
@@ -209,6 +256,7 @@ public final class ZkController {
});
+
this.overseerJobQueue = Overseer.getInQueue(zkClient);
this.overseerCollectionQueue = Overseer.getCollectionQueue(zkClient);
cmdExecutor = new ZkCmdExecutor();
@@ -218,8 +266,12 @@ public final class ZkController {
init(registerOnReconnect);
}
+ public String getLeaderVoteWait() {
+ return leaderVoteWait;
+ }
+
private void registerAllCoresAsDown(
- final CurrentCoreDescriptorProvider registerOnReconnect) {
+ final CurrentCoreDescriptorProvider registerOnReconnect, boolean updateLastPublished) {
List<CoreDescriptor> descriptors = registerOnReconnect
.getCurrentDescriptors();
if (descriptors != null) {
@@ -229,33 +281,69 @@ public final class ZkController {
final String coreZkNodeName = getNodeName() + "_"
+ descriptor.getName();
try {
- publish(descriptor, ZkStateReader.DOWN);
+ descriptor.getCloudDescriptor().isLeader = false;
+ publish(descriptor, ZkStateReader.DOWN, updateLastPublished);
+ } catch (Exception e) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e1) {
+ Thread.currentThread().interrupt();
+ }
+ try {
+ publish(descriptor, ZkStateReader.DOWN);
+ } catch (Exception e2) {
+ SolrException.log(log, "", e2);
+ continue;
+ }
+ }
+ try {
waitForLeaderToSeeDownState(descriptor, coreZkNodeName);
} catch (Exception e) {
SolrException.log(log, "", e);
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e1) {
+ Thread.currentThread().interrupt();
+ }
}
}
}
}
+
+ private void markAllAsNotLeader(
+ final CurrentCoreDescriptorProvider registerOnReconnect) {
+ List<CoreDescriptor> descriptors = registerOnReconnect
+ .getCurrentDescriptors();
+ if (descriptors != null) {
+ for (CoreDescriptor descriptor : descriptors) {
+ descriptor.getCloudDescriptor().isLeader = false;
+ }
+ }
+ }
/**
* Closes the underlying ZooKeeper client.
*/
public void close() {
+
+ if (cmdDistribExecutor != null) {
+ try {
+ ExecutorUtil.shutdownNowAndAwaitTermination(cmdDistribExecutor);
+ } catch (Throwable e) {
+ SolrException.log(log, e);
+ }
+ }
+
+ for (ElectionContext context : electionContexts.values()) {
+ context.close();
+ }
+
try {
overseer.close();
} catch(Throwable t) {
log.error("Error closing overseer", t);
}
- try {
- zkClient.close();
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.warn("", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR,
- "", e);
- }
+ zkClient.close();
}
/**
@@ -351,7 +439,7 @@ public final class ZkController {
}
private void init(CurrentCoreDescriptorProvider registerOnReconnect) {
- registerAllCoresAsDown(registerOnReconnect);
+ registerAllCoresAsDown(registerOnReconnect, true);
try {
// makes nodes zkNode
@@ -362,13 +450,8 @@ public final class ZkController {
ShardHandler shardHandler;
String adminPath;
- if (cc == null) {
- shardHandler = new HttpShardHandlerFactory().getShardHandler();
- adminPath = "/admin/cores";
- } else {
- shardHandler = cc.getShardHandlerFactory().getShardHandler();
- adminPath = cc.getAdminPath();
- }
+ shardHandler = cc.getShardHandlerFactory().getShardHandler();
+ adminPath = cc.getAdminPath();
overseerElector = new LeaderElector(zkClient);
this.overseer = new Overseer(shardHandler, adminPath, zkStateReader);
@@ -467,7 +550,7 @@ public final class ZkController {
if(data != null) {
ZkNodeProps props = ZkNodeProps.load(data);
- configName = props.get(CONFIGNAME_PROP);
+ configName = props.getStr(CONFIGNAME_PROP);
}
if (configName != null && !zkClient.exists(CONFIGS_ZKNODE + "/" + configName, true)) {
@@ -490,7 +573,7 @@ public final class ZkController {
* @throws Exception
*/
public String register(String coreName, final CoreDescriptor desc) throws Exception {
- return register(coreName, desc, false);
+ return register(coreName, desc, false, false);
}
@@ -500,10 +583,11 @@ public final class ZkController {
* @param coreName
* @param desc
* @param recoverReloadedCores
+ * @param afterExpiration
* @return the shardId for the SolrCore
* @throws Exception
*/
- public String register(String coreName, final CoreDescriptor desc, boolean recoverReloadedCores) throws Exception {
+ public String register(String coreName, final CoreDescriptor desc, boolean recoverReloadedCores, boolean afterExpiration) throws Exception {
final String baseUrl = getBaseUrl();
final CloudDescriptor cloudDesc = desc.getCloudDescriptor();
@@ -513,7 +597,7 @@ public final class ZkController {
String shardId = cloudDesc.getShardId();
- Map<String,String> props = new HashMap<String,String>();
+ Map<String,Object> props = new HashMap<String,Object>();
// we only put a subset of props into the leader node
props.put(ZkStateReader.BASE_URL_PROP, baseUrl);
props.put(ZkStateReader.CORE_NAME_PROP, coreName);
@@ -528,7 +612,7 @@ public final class ZkController {
ZkNodeProps leaderProps = new ZkNodeProps(props);
try {
- joinElection(desc);
+ joinElection(desc, afterExpiration);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -539,25 +623,7 @@ public final class ZkController {
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
}
- // rather than look in the cluster state file, we go straight to the zknodes
- // here, because on cluster restart there could be stale leader info in the
- // cluster state node that won't be updated for a moment
- String leaderUrl = getLeaderProps(collection, cloudDesc.getShardId()).getCoreUrl();
-
- // now wait until our currently cloud state contains the latest leader
- String clusterStateLeader = zkStateReader.getLeaderUrl(collection, shardId, 30000);
- int tries = 0;
- while (!leaderUrl.equals(clusterStateLeader)) {
- if (tries == 60) {
- throw new SolrException(ErrorCode.SERVER_ERROR,
- "There is conflicting information about the leader of shard: "
- + cloudDesc.getShardId() + " our state says:" + clusterStateLeader + " but zookeeper says:" + leaderUrl);
- }
- Thread.sleep(1000);
- tries++;
- clusterStateLeader = zkStateReader.getLeaderUrl(collection, shardId, 30000);
- leaderUrl = getLeaderProps(collection, cloudDesc.getShardId()).getCoreUrl();
- }
+ String leaderUrl = getLeader(cloudDesc);
String ourUrl = ZkCoreNodeProps.getCoreUrl(baseUrl, coreName);
log.info("We are " + ourUrl + " and leader is " + leaderUrl);
@@ -565,66 +631,109 @@ public final class ZkController {
SolrCore core = null;
- if (cc != null) { // CoreContainer only null in tests
- try {
- core = cc.getCore(desc.getName());
+ try {
+ core = cc.getCore(desc.getName());
- // recover from local transaction log and wait for it to complete before
- // going active
- // TODO: should this be moved to another thread? To recoveryStrat?
- // TODO: should this actually be done earlier, before (or as part of)
- // leader election perhaps?
- // TODO: if I'm the leader, ensure that a replica that is trying to recover waits until I'm
- // active (or don't make me the
- // leader until my local replay is done.
-
- UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
- if (!core.isReloaded() && ulog != null) {
- Future<UpdateLog.RecoveryInfo> recoveryFuture = core.getUpdateHandler()
- .getUpdateLog().recoverFromLog();
- if (recoveryFuture != null) {
- recoveryFuture.get(); // NOTE: this could potentially block for
- // minutes or more!
- // TODO: public as recovering in the mean time?
- // TODO: in the future we could do peerync in parallel with recoverFromLog
- } else {
- log.info("No LogReplay needed for core="+core.getName() + " baseURL=" + baseUrl);
- }
- }
-
- boolean didRecovery = checkRecovery(coreName, desc, recoverReloadedCores, isLeader, cloudDesc,
- collection, coreZkNodeName, shardId, leaderProps, core, cc);
- if (!didRecovery) {
- publish(desc, ZkStateReader.ACTIVE);
- }
- } finally {
- if (core != null) {
- core.close();
+ // recover from local transaction log and wait for it to complete before
+ // going active
+ // TODO: should this be moved to another thread? To recoveryStrat?
+ // TODO: should this actually be done earlier, before (or as part of)
+ // leader election perhaps?
+ // TODO: if I'm the leader, ensure that a replica that is trying to recover waits until I'm
+ // active (or don't make me the
+ // leader until my local replay is done.
+
+ UpdateLog ulog = core.getUpdateHandler().getUpdateLog();
+ if (!core.isReloaded() && ulog != null) {
+ Future<UpdateLog.RecoveryInfo> recoveryFuture = core.getUpdateHandler()
+ .getUpdateLog().recoverFromLog();
+ if (recoveryFuture != null) {
+ recoveryFuture.get(); // NOTE: this could potentially block for
+ // minutes or more!
+ // TODO: public as recovering in the mean time?
+ // TODO: in the future we could do peerync in parallel with recoverFromLog
+ } else {
+ log.info("No LogReplay needed for core="+core.getName() + " baseURL=" + baseUrl);
}
+ }
+ boolean didRecovery = checkRecovery(coreName, desc, recoverReloadedCores, isLeader, cloudDesc,
+ collection, coreZkNodeName, shardId, leaderProps, core, cc);
+ if (!didRecovery) {
+ publish(desc, ZkStateReader.ACTIVE);
+ }
+ } finally {
+ if (core != null) {
+ core.close();
}
- } else {
- publish(desc, ZkStateReader.ACTIVE);
}
+
// make sure we have an update cluster state right away
zkStateReader.updateClusterState(true);
return shardId;
}
+
+ private String getLeader(final CloudDescriptor cloudDesc) {
+
+ String collection = cloudDesc.getCollectionName();
+ String shardId = cloudDesc.getShardId();
+ // rather than look in the cluster state file, we go straight to the zknodes
+ // here, because on cluster restart there could be stale leader info in the
+ // cluster state node that won't be updated for a moment
+ String leaderUrl;
+ try {
+ leaderUrl = getLeaderProps(collection, cloudDesc.getShardId())
+ .getCoreUrl();
+
+ // now wait until our currently cloud state contains the latest leader
+ String clusterStateLeader = zkStateReader.getLeaderUrl(collection,
+ shardId, 30000);
+ int tries = 0;
+ while (!leaderUrl.equals(clusterStateLeader)) {
+ if (tries == 60) {
+ throw new SolrException(ErrorCode.SERVER_ERROR,
+ "There is conflicting information about the leader of shard: "
+ + cloudDesc.getShardId() + " our state says:"
+ + clusterStateLeader + " but zookeeper says:" + leaderUrl);
+ }
+ Thread.sleep(1000);
+ tries++;
+ clusterStateLeader = zkStateReader.getLeaderUrl(collection, shardId,
+ 30000);
+ leaderUrl = getLeaderProps(collection, cloudDesc.getShardId())
+ .getCoreUrl();
+ }
+
+ } catch (Exception e) {
+ log.error("Error getting leader from zk", e);
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
+ "Error getting leader from zk", e);
+ }
+ return leaderUrl;
+ }
+
+ /**
+ * Get leader props directly from zk nodes.
+ */
+ public ZkCoreNodeProps getLeaderProps(final String collection,
+ final String slice) throws InterruptedException {
+ return getLeaderProps(collection, slice, false);
+ }
/**
* Get leader props directly from zk nodes.
*
* @param collection
* @param slice
- * @return
- * @throws KeeperException
+ * @return leader props
* @throws InterruptedException
*/
- private ZkCoreNodeProps getLeaderProps(final String collection,
- final String slice) throws KeeperException, InterruptedException {
+ public ZkCoreNodeProps getLeaderProps(final String collection,
+ final String slice, boolean failImmediatelyOnExpiration) throws InterruptedException {
int iterCount = 60;
+ Exception exp = null;
while (iterCount-- > 0) {
try {
byte[] data = zkClient.getData(
@@ -633,19 +742,31 @@ public final class ZkController {
ZkCoreNodeProps leaderProps = new ZkCoreNodeProps(
ZkNodeProps.load(data));
return leaderProps;
- } catch (NoNodeException e) {
+ } catch (InterruptedException e) {
+ throw e;
+ } catch (SessionExpiredException e) {
+ if (failImmediatelyOnExpiration) {
+ throw new RuntimeException("Session has expired - could not get leader props", exp);
+ }
+ exp = e;
+ Thread.sleep(500);
+ } catch (Exception e) {
+ exp = e;
Thread.sleep(500);
}
+ if (cc.isShutDown()) {
+ throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "CoreContainer is shutdown");
+ }
}
- throw new RuntimeException("Could not get leader props");
+ throw new SolrException(ErrorCode.SERVICE_UNAVAILABLE, "Could not get leader props", exp);
}
- private void joinElection(CoreDescriptor cd) throws InterruptedException, KeeperException, IOException {
+ private void joinElection(CoreDescriptor cd, boolean afterExpiration) throws InterruptedException, KeeperException, IOException {
String shardId = cd.getCloudDescriptor().getShardId();
- Map<String,String> props = new HashMap<String,String>();
+ Map<String,Object> props = new HashMap<String,Object>();
// we only put a subset of props into the leader node
props.put(ZkStateReader.BASE_URL_PROP, getBaseUrl());
props.put(ZkStateReader.CORE_NAME_PROP, cd.getName());
@@ -712,6 +833,10 @@ public final class ZkController {
return baseURL;
}
+ public void publish(final CoreDescriptor cd, final String state) throws KeeperException, InterruptedException {
+ publish(cd, state, true);
+ }
+
/**
* Publish core state to overseer.
* @param cd
@@ -719,7 +844,7 @@ public final class ZkController {
* @throws KeeperException
* @throws InterruptedException
*/
- public void publish(final CoreDescriptor cd, final String state) throws KeeperException, InterruptedException {
+ public void publish(final CoreDescriptor cd, final String state, boolean updateLastState) throws KeeperException, InterruptedException {
//System.out.println(Thread.currentThread().getStackTrace()[3]);
Integer numShards = cd.getCloudDescriptor().getNumShards();
if (numShards == null) { //XXX sys prop hack
@@ -737,6 +862,7 @@ public final class ZkController {
.getCollectionName(), ZkStateReader.STATE_PROP, state,
ZkStateReader.NUM_SHARDS_PROP, numShards != null ? numShards.toString()
: null);
+ cd.getCloudDescriptor().lastPublished = state;
overseerJobQueue.offer(ZkStateReader.toJSON(m));
}
@@ -822,7 +948,7 @@ public final class ZkController {
SolrParams params = cd.getParams();
try {
- Map<String,String> collectionProps = new HashMap<String,String>();
+ Map<String,Object> collectionProps = new HashMap<String,Object>();
// TODO: if collection.configName isn't set, and there isn't already a conf in zk, just use that?
String defaultConfigName = System.getProperty(COLLECTION_PARAM_PREFIX+CONFIGNAME_PROP, collection);
@@ -887,7 +1013,7 @@ public final class ZkController {
private void getConfName(String collection, String collectionPath,
- Map<String,String> collectionProps) throws KeeperException,
+ Map<String,Object> collectionProps) throws KeeperException,
InterruptedException {
// check for configName
log.info("Looking for collection configName");
@@ -1060,7 +1186,6 @@ public final class ZkController {
prepCmd.setNodeName(getNodeName());
prepCmd.setCoreNodeName(coreZkNodeName);
prepCmd.setState(ZkStateReader.DOWN);
- prepCmd.setPauseFor(0);
// let's retry a couple times - perhaps the leader just went down,
// or perhaps he is just not quite ready for us yet
@@ -1118,7 +1243,7 @@ public final class ZkController {
ZkNodeProps props = null;
if(data != null) {
props = ZkNodeProps.load(data);
- Map<String,String> newProps = new HashMap<String,String>();
+ Map<String,Object> newProps = new HashMap<String,Object>();
newProps.putAll(props.getProperties());
newProps.put(CONFIGNAME_PROP, confSetName);
props = new ZkNodeProps(newProps);
@@ -1168,5 +1293,14 @@ public final class ZkController {
public DistributedQueue getOverseerCollectionQueue() {
return overseerCollectionQueue;
}
+
+ public int getClientTimeout() {
+ return clientTimeout;
+ }
+
+ // may return null if not in zk mode
+ public ThreadPoolExecutor getCmdDistribExecutor() {
+ return cmdDistribExecutor;
+ }
}
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CachingDirectoryFactory.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CachingDirectoryFactory.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CachingDirectoryFactory.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CachingDirectoryFactory.java Fri Sep 21 17:21:34 2012
@@ -47,6 +47,9 @@ public abstract class CachingDirectoryFa
int refCnt = 1;
public String path;
public boolean doneWithDir = false;
+ public String toString() {
+ return "CachedDir<<" + directory.toString() + ";refCount=" + refCnt + ";path=" + path + ";done=" + doneWithDir + ">>";
+ }
}
private static Logger log = LoggerFactory
@@ -105,7 +108,11 @@ public abstract class CachingDirectoryFa
public void close() throws IOException {
synchronized (this) {
for (CacheValue val : byDirectoryCache.values()) {
- val.directory.close();
+ try {
+ val.directory.close();
+ } catch (Throwable t) {
+ SolrException.log(log, "Error closing directory", t);
+ }
}
byDirectoryCache.clear();
byPathCache.clear();
@@ -119,8 +126,12 @@ public abstract class CachingDirectoryFa
throw new IllegalArgumentException("Unknown directory: " + directory
+ " " + byDirectoryCache);
}
+
+ log.debug("Closing: {}", cacheValue);
+
cacheValue.refCnt--;
if (cacheValue.refCnt == 0 && cacheValue.doneWithDir) {
+ log.info("Closing directory:" + cacheValue.path);
directory.close();
byDirectoryCache.remove(directory);
byPathCache.remove(cacheValue.path);
@@ -190,6 +201,7 @@ public abstract class CachingDirectoryFa
byDirectoryCache.put(directory, newCacheValue);
byPathCache.put(fullPath, newCacheValue);
+ log.info("return new directory for " + fullPath + " forceNew:" + forceNew);
} else {
cacheValue.refCnt++;
}
@@ -233,13 +245,6 @@ public abstract class CachingDirectoryFa
close(directory);
}
- /**
- * @param dir
- * @param lockPath
- * @param rawLockType
- * @return
- * @throws IOException
- */
private static Directory injectLockFactory(Directory dir, String lockPath,
String rawLockType) throws IOException {
if (null == rawLockType) {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/Config.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/Config.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/Config.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/Config.java Fri Sep 21 17:21:34 2012
@@ -141,8 +141,8 @@ public class Config {
SolrException.log(log, "Exception during parsing file: " + name, e);
throw e;
} catch( SolrException e ){
- SolrException.log(log,"Error in "+name,e);
- throw e;
+ SolrException.log(log,"Error in "+name,e);
+ throw e;
}
}
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CoreContainer.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CoreContainer.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CoreContainer.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/CoreContainer.java Fri Sep 21 17:21:34 2012
@@ -34,6 +34,9 @@ import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.xml.parsers.ParserConfigurationException;
@@ -56,10 +59,7 @@ import org.apache.solr.common.SolrExcept
import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.ZooKeeperException;
-import org.apache.solr.common.params.CoreAdminParams;
-import org.apache.solr.util.DOMUtil;
-import org.apache.solr.util.FileUtils;
-import org.apache.solr.util.SystemIdResolver;
+import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.core.SolrXMLSerializer.SolrCoreXMLDef;
import org.apache.solr.core.SolrXMLSerializer.SolrXMLDef;
import org.apache.solr.handler.admin.CollectionsHandler;
@@ -71,6 +71,10 @@ import org.apache.solr.logging.LogWatche
import org.apache.solr.logging.jul.JulWatcher;
import org.apache.solr.schema.IndexSchema;
import org.apache.solr.update.SolrCoreState;
+import org.apache.solr.util.DOMUtil;
+import org.apache.solr.util.DefaultSolrThreadFactory;
+import org.apache.solr.util.FileUtils;
+import org.apache.solr.util.SystemIdResolver;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -88,6 +92,7 @@ import org.xml.sax.SAXException;
*/
public class CoreContainer
{
+ private static final String LEADER_VOTE_WAIT = "180000"; // 3 minutes
private static final String DEFAULT_HOST_CONTEXT = "solr";
private static final String DEFAULT_HOST_PORT = "8983";
private static final int DEFAULT_ZK_CLIENT_TIMEOUT = 15000;
@@ -139,8 +144,8 @@ public class CoreContainer
protected LogWatcher logging = null;
private String zkHost;
private Map<SolrCore,String> coreToOrigName = new ConcurrentHashMap<SolrCore,String>();
-
-
+ private String leaderVoteWait;
+
{
log.info("New CoreContainer " + System.identityHashCode(this));
}
@@ -227,7 +232,7 @@ public class CoreContainer
} else {
log.info("Zookeeper client=" + zookeeperHost);
}
- zkController = new ZkController(this, zookeeperHost, zkClientTimeout, zkClientConnectTimeout, host, hostPort, hostContext, new CurrentCoreDescriptorProvider() {
+ zkController = new ZkController(this, zookeeperHost, zkClientTimeout, zkClientConnectTimeout, host, hostPort, hostContext, leaderVoteWait, new CurrentCoreDescriptorProvider() {
@Override
public List<CoreDescriptor> getCurrentDescriptors() {
@@ -456,6 +461,8 @@ public class CoreContainer
hostContext = cfg.get("solr/cores/@hostContext", DEFAULT_HOST_CONTEXT);
host = cfg.get("solr/cores/@host", null);
+
+ leaderVoteWait = cfg.get("solr/cores/@leaderVoteWait", LEADER_VOTE_WAIT);
if(shareSchema){
indexSchemaCache = new ConcurrentHashMap<String ,IndexSchema>();
@@ -522,7 +529,7 @@ public class CoreContainer
}
opt = DOMUtil.getAttr(node, CORE_ROLES, null);
if(opt != null){
- p.getCloudDescriptor().setRoles(opt);
+ p.getCloudDescriptor().setRoles(opt);
}
}
opt = DOMUtil.getAttr(node, CORE_PROPERTIES, null);
@@ -580,38 +587,42 @@ public class CoreContainer
* Stops all cores.
*/
public void shutdown() {
- log.info("Shutting down CoreContainer instance="+System.identityHashCode(this));
+ log.info("Shutting down CoreContainer instance="
+ + System.identityHashCode(this));
isShutDown = true;
if (isZooKeeperAware()) {
cancelCoreRecoveries();
}
-
- synchronized(cores) {
- try {
+ try {
+ synchronized (cores) {
+
for (SolrCore core : cores.values()) {
try {
- core.close();
+ core.close();
} catch (Throwable t) {
SolrException.log(log, "Error shutting down core", t);
}
}
cores.clear();
- } finally {
- if(zkController != null) {
- zkController.close();
- }
- if (zkServer != null) {
- zkServer.stop();
- }
- if (shardHandlerFactory != null) {
- shardHandlerFactory.close();
- }
}
+ } finally {
+ if (shardHandlerFactory != null) {
+ shardHandlerFactory.close();
+ }
+
+ // we want to close zk stuff last
+ if (zkController != null) {
+ zkController.close();
+ }
+ if (zkServer != null) {
+ zkServer.stop();
+ }
+
}
}
- private void cancelCoreRecoveries() {
+ public void cancelCoreRecoveries() {
ArrayList<SolrCoreState> coreStates = new ArrayList<SolrCoreState>();
synchronized (cores) {
for (SolrCore core : cores.values()) {
@@ -806,7 +817,7 @@ public class CoreContainer
IndexSchema schema = null;
if (indexSchemaCache != null) {
- if (zkController != null) {
+ if (zkController == null) {
File schemaFile = new File(dcore.getSchemaName());
if (!schemaFile.isAbsolute()) {
schemaFile = new File(solrLoader.getInstanceDir() + "conf"
@@ -875,6 +886,9 @@ public class CoreContainer
failure = e4;
throw e4;
} finally {
+ if (null != failure) {
+ log.error("Unable to create core: " + name, failure);
+ }
synchronized (coreInitFailures) {
// remove first so insertion order is updated and newest is last
coreInitFailures.remove(name);
@@ -1035,6 +1049,9 @@ public class CoreContainer
failure = e4;
throw e4;
} finally {
+ if (null != failure) {
+ log.error("Unable to reload core: " + name, failure);
+ }
synchronized (coreInitFailures) {
// remove first so insertion order is updated and newest is last
coreInitFailures.remove(name);
@@ -1230,6 +1247,7 @@ public class CoreContainer
intToString(this.zkClientTimeout),
Integer.toString(DEFAULT_ZK_CLIENT_TIMEOUT));
addCoresAttrib(coresAttribs, "hostContext", this.hostContext, DEFAULT_HOST_CONTEXT);
+ addCoresAttrib(coresAttribs, "leaderVoteWait", this.leaderVoteWait, LEADER_VOTE_WAIT);
List<SolrCoreXMLDef> solrCoreXMLDefs = new ArrayList<SolrCoreXMLDef>();
@@ -1302,11 +1320,15 @@ public class CoreContainer
CloudDescriptor cd = dcore.getCloudDescriptor();
String shard = null;
+ String roles = null;
if (cd != null) {
shard = cd.getShardId();
+ roles = cd.getRoles();
}
addCoreProperty(coreAttribs, coreNode, CORE_SHARD, shard, null);
+ addCoreProperty(coreAttribs, coreNode, CORE_ROLES, roles, null);
+
String collection = null;
// only write out the collection name if it's not the default (the
// core
@@ -1398,6 +1420,10 @@ public class CoreContainer
public ZkController getZkController() {
return zkController;
}
+
+ public boolean isShareSchema() {
+ return shareSchema;
+ }
/** The default ShardHandlerFactory used to communicate with other solr instances */
public ShardHandlerFactory getShardHandlerFactory() {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/MMapDirectoryFactory.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/MMapDirectoryFactory.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/MMapDirectoryFactory.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/MMapDirectoryFactory.java Fri Sep 21 17:21:34 2012
@@ -18,6 +18,7 @@ package org.apache.solr.core;
import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.LockFactory; // javadocs
import org.apache.lucene.store.MMapDirectory;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;
@@ -33,8 +34,8 @@ import java.io.IOException;
*
* Can set the following parameters:
* <ul>
- * <li>unmap -- See {@link org.apache.lucene.store.MMapDirectory#setUseUnmap(boolean)}</li>
- * <li>maxChunkSize -- The Max chunk size. See {@link org.apache.lucene.store.MMapDirectory#setMaxChunkSize(int)}</li>
+ * <li>unmap -- See {@link MMapDirectory#setUseUnmap(boolean)}</li>
+ * <li>maxChunkSize -- The Max chunk size. See {@link MMapDirectory#MMapDirectory(File, LockFactory, int)}</li>
* </ul>
*
**/
@@ -55,13 +56,12 @@ public class MMapDirectoryFactory extend
@Override
protected Directory create(String path) throws IOException {
- MMapDirectory mapDirectory = new MMapDirectory(new File(path));
+ MMapDirectory mapDirectory = new MMapDirectory(new File(path), null, maxChunk);
try {
mapDirectory.setUseUnmap(unmapHack);
} catch (Exception e) {
log.warn("Unmap not supported on this JVM, continuing on without setting unmap", e);
}
- mapDirectory.setMaxChunkSize(maxChunk);
return mapDirectory;
}
}
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrCore.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrCore.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrCore.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrCore.java Fri Sep 21 17:21:34 2012
@@ -65,6 +65,7 @@ import org.apache.solr.search.ValueSourc
import org.apache.solr.update.DirectUpdateHandler2;
import org.apache.solr.update.SolrIndexWriter;
import org.apache.solr.update.UpdateHandler;
+import org.apache.solr.update.VersionInfo;
import org.apache.solr.update.processor.DistributedUpdateProcessorFactory;
import org.apache.solr.update.processor.LogUpdateProcessorFactory;
import org.apache.solr.update.processor.RunUpdateProcessorFactory;
@@ -409,7 +410,7 @@ public final class SolrCore implements S
// protect via synchronized(SolrCore.class)
private static Set<String> dirs = new HashSet<String>();
- void initIndex() {
+ void initIndex(boolean reload) {
try {
String indexDir = getNewIndexDir();
boolean indexExists = getDirectoryFactory().exists(indexDir);
@@ -421,7 +422,7 @@ public final class SolrCore implements S
initIndexReaderFactory();
- if (indexExists && firstTime) {
+ if (indexExists && firstTime && !reload) {
// to remove locks, the directory must already exist... so we create it
// if it didn't exist already...
Directory dir = directoryFactory.get(indexDir, getSolrConfig().indexConfig.lockType);
@@ -444,7 +445,7 @@ public final class SolrCore implements S
log.warn(logid+"Solr index directory '" + new File(indexDir) + "' doesn't exist."
+ " Creating new index...");
- SolrIndexWriter writer = new SolrIndexWriter("SolrCore.initIndex", indexDir, getDirectoryFactory(), true, schema, solrConfig.indexConfig, solrDelPolicy, codec, false);
+ SolrIndexWriter writer = SolrIndexWriter.create("SolrCore.initIndex", indexDir, getDirectoryFactory(), true, schema, solrConfig.indexConfig, solrDelPolicy, codec, false);
writer.close();
}
@@ -479,6 +480,13 @@ public final class SolrCore implements S
} catch (SolrException e) {
throw e;
} catch (Exception e) {
+ // The JVM likes to wrap our helpful SolrExceptions in things like
+ // "InvocationTargetException" that have no useful getMessage
+ if (null != e.getCause() && e.getCause() instanceof SolrException) {
+ SolrException inner = (SolrException) e.getCause();
+ throw inner;
+ }
+
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,"Error Instantiating "+msg+", "+className+ " failed to instantiate " +cast.getName(), e);
}
}
@@ -502,6 +510,13 @@ public final class SolrCore implements S
} catch (SolrException e) {
throw e;
} catch (Exception e) {
+ // The JVM likes to wrap our helpful SolrExceptions in things like
+ // "InvocationTargetException" that have no useful getMessage
+ if (null != e.getCause() && e.getCause() instanceof SolrException) {
+ SolrException inner = (SolrException) e.getCause();
+ throw inner;
+ }
+
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,"Error Instantiating "+msg+", "+className+ " failed to instantiate " + UpdateHandler.class.getName(), e);
}
}
@@ -579,6 +594,25 @@ public final class SolrCore implements S
schema = new IndexSchema(config, IndexSchema.DEFAULT_SCHEMA_FILE, null);
}
+ if (null != cd && null != cd.getCloudDescriptor()) {
+ // we are evidently running in cloud mode.
+ //
+ // In cloud mode, version field is required for correct consistency
+ // ideally this check would be more fine grained, and individual features
+ // would assert it when they initialize, but DistribuedUpdateProcessor
+ // is currently a big ball of wax that does more then just distributing
+ // updates (ie: partial document updates), so it needs to work in no cloud
+ // mode as well, and can't assert version field support on init.
+
+ try {
+ Object ignored = VersionInfo.getAndCheckVersionField(schema);
+ } catch (SolrException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
+ "Schema will not work with SolrCloud mode: " +
+ e.getMessage(), e);
+ }
+ }
+
//Initialize JMX
if (config.jmxConfig.enabled) {
infoRegistry = new JmxMonitoredMap<String, SolrInfoMBean>(name, String.valueOf(this.hashCode()), config.jmxConfig);
@@ -610,7 +644,7 @@ public final class SolrCore implements S
this.isReloaded = true;
}
- initIndex();
+ initIndex(prev != null);
initWriters();
initQParsers();
@@ -775,7 +809,8 @@ public final class SolrCore implements S
// this core current usage count
private final AtomicInteger refCount = new AtomicInteger(1);
- final void open() {
+ /** expert: increments the core reference count */
+ public void open() {
refCount.incrementAndGet();
}
@@ -791,7 +826,7 @@ public final class SolrCore implements S
* <p>
* <p>
* The behavior of this method is determined by the result of decrementing
- * the core's reference count (A core is created with a refrence count of 1)...
+ * the core's reference count (A core is created with a reference count of 1)...
* </p>
* <ul>
* <li>If reference count is > 0, the usage count is decreased by 1 and no
@@ -862,7 +897,15 @@ public final class SolrCore implements S
}
try {
- if (updateHandler != null) updateHandler.close();
+ if (null != updateHandler) {
+ updateHandler.close();
+ } else {
+ if (null != directoryFactory) {
+ // :HACK: normally we rely on updateHandler to do this,
+ // but what if updateHandler failed to init?
+ directoryFactory.close();
+ }
+ }
} catch (Throwable e) {
SolrException.log(log,e);
}
@@ -948,14 +991,14 @@ public final class SolrCore implements S
}
/**
- * Returns an unmodifieable Map containing the registered handlers of the specified type.
+ * Returns an unmodifiable Map containing the registered handlers of the specified type.
*/
public Map<String,SolrRequestHandler> getRequestHandlers(Class clazz) {
return reqHandlers.getAll(clazz);
}
/**
- * Returns an unmodifieable Map containing the registered handlers
+ * Returns an unmodifiable Map containing the registered handlers
*/
public Map<String,SolrRequestHandler> getRequestHandlers() {
return reqHandlers.getRequestHandlers();
@@ -973,8 +1016,8 @@ public final class SolrCore implements S
* http://${host}:${port}/${context}/select?qt=${handlerName}
* </pre>
*
- * Handlers <em>must</em> be initalized before getting registered. Registered
- * handlers can immediatly accept requests.
+ * Handlers <em>must</em> be initialized before getting registered. Registered
+ * handlers can immediately accept requests.
*
* This call is thread safe.
*
@@ -1160,7 +1203,7 @@ public final class SolrCore implements S
}
- /** Opens a new searcher and returns a RefCounted<SolrIndexSearcher> with it's reference incremented.
+ /** Opens a new searcher and returns a RefCounted<SolrIndexSearcher> with it's reference incremented.
*
* "realtime" means that we need to open quickly for a realtime view of the index, hence don't do any
* autowarming and add to the _realtimeSearchers queue rather than the _searchers queue (so it won't
@@ -1169,7 +1212,7 @@ public final class SolrCore implements S
*
* realtimeSearcher is updated to the latest opened searcher, regardless of the value of "realtime".
*
- * This method aquires openSearcherLock - do not call with searckLock held!
+ * This method acquires openSearcherLock - do not call with searckLock held!
*/
public RefCounted<SolrIndexSearcher> openNewSearcher(boolean updateHandlerReopens, boolean realtime) {
SolrIndexSearcher tmp;
@@ -1554,7 +1597,7 @@ public final class SolrCore implements S
} catch (Throwable e) {
// do not allow decref() operations to fail since they are typically called in finally blocks
// and throwing another exception would be very unexpected.
- SolrException.log(log, "Error closing searcher:", e);
+ SolrException.log(log, "Error closing searcher:" + this, e);
}
}
};
@@ -1941,7 +1984,7 @@ public final class SolrCore implements S
/**
*
* @param registry The map to which the instance should be added to. The key is the name attribute
- * @param type The type of the Plugin. These should be standard ones registerd by type.getName() in SolrConfig
+ * @param type The type of the Plugin. These should be standard ones registered by type.getName() in SolrConfig
* @return The default if any
*/
public <T> T initPlugins(Map<String, T> registry, Class<T> type) {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrResourceLoader.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrResourceLoader.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrResourceLoader.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/core/SolrResourceLoader.java Fri Sep 21 17:21:34 2012
@@ -36,7 +36,6 @@ import org.apache.lucene.analysis.util.T
import org.apache.lucene.analysis.util.TokenizerFactory;
import org.apache.lucene.codecs.Codec;
import org.apache.lucene.codecs.PostingsFormat;
-import org.apache.lucene.util.hash.HashFunction;
import org.apache.lucene.analysis.util.WordlistLoader;
import org.apache.solr.common.ResourceLoader;
import org.apache.solr.handler.admin.CoreAdminHandler;
@@ -176,8 +175,6 @@ public class SolrResourceLoader implemen
* this ResourceLoader.
*/
void reloadLuceneSPI() {
- // Hash functions:
- HashFunction.reloadHashFunctions(this.classLoader);
// Codecs:
PostingsFormat.reloadPostingsFormats(this.classLoader);
Codec.reloadCodecs(this.classLoader);
@@ -634,7 +631,11 @@ public class SolrResourceLoader implemen
for (SolrInfoMBean bean : arr) {
- infoRegistry.put(bean.getName(), bean);
+ try {
+ infoRegistry.put(bean.getName(), bean);
+ } catch (Throwable t) {
+ log.warn("could not register MBean '" + bean.getName() + "'.", t);
+ }
}
}
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/CSVRequestHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/CSVRequestHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/CSVRequestHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/CSVRequestHandler.java Fri Sep 21 17:21:34 2012
@@ -29,7 +29,7 @@ public class CSVRequestHandler extends U
public void init(NamedList args) {
super.init(args);
setAssumeContentType("application/csv");
- log.warn("Using deprecated class: "+this.getClass().getSimpleName()+" -- replace with UpdateRequestHandler");
+ // log.warn("Using deprecated class: "+this.getClass().getSimpleName()+" -- replace with UpdateRequestHandler");
}
//////////////////////// SolrInfoMBeans methods //////////////////////
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/JsonUpdateRequestHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/JsonUpdateRequestHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/JsonUpdateRequestHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/JsonUpdateRequestHandler.java Fri Sep 21 17:21:34 2012
@@ -29,7 +29,7 @@ public class JsonUpdateRequestHandler ex
public void init(NamedList args) {
super.init(args);
setAssumeContentType("application/json");
- log.warn("Using deprecated class: "+this.getClass().getSimpleName()+" -- replace with UpdateRequestHandler");
+ // log.warn("Using deprecated class: "+this.getClass().getSimpleName()+" -- replace with UpdateRequestHandler");
}
//////////////////////// SolrInfoMBeans methods //////////////////////
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/MoreLikeThisHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/MoreLikeThisHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/MoreLikeThisHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/MoreLikeThisHandler.java Fri Sep 21 17:21:34 2012
@@ -31,6 +31,7 @@ import java.util.regex.Pattern;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.StoredDocument;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryparser.classic.ParseException;
import org.apache.lucene.search.*;
@@ -342,7 +343,7 @@ public class MoreLikeThisHandler extends
public DocListAndSet getMoreLikeThis( int id, int start, int rows, List<Query> filters, List<InterestingTerm> terms, int flags ) throws IOException
{
- Document doc = reader.document(id);
+ StoredDocument doc = reader.document(id);
rawMLTQuery = mlt.like(id);
boostedMLTQuery = getBoostedQuery( rawMLTQuery );
if( terms != null ) {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java Fri Sep 21 17:21:34 2012
@@ -283,7 +283,7 @@ public class ReplicationHandler extends
private volatile SnapPuller tempSnapPuller;
- public boolean doFetch(SolrParams solrParams, boolean force) {
+ public boolean doFetch(SolrParams solrParams, boolean forceReplication) {
String masterUrl = solrParams == null ? null : solrParams.get(MASTER_URL);
if (!snapPullLock.tryLock())
return false;
@@ -294,7 +294,7 @@ public class ReplicationHandler extends
nl.remove(SnapPuller.POLL_INTERVAL);
tempSnapPuller = new SnapPuller(nl, this, core);
}
- return tempSnapPuller.fetchLatestIndex(core, force);
+ return tempSnapPuller.fetchLatestIndex(core, forceReplication);
} catch (Exception e) {
SolrException.log(LOG, "SnapPull failed ", e);
} finally {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/SnapPuller.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/SnapPuller.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/SnapPuller.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/SnapPuller.java Fri Sep 21 17:21:34 2012
@@ -243,12 +243,11 @@ public class SnapPuller {
* downloaded. It also downloads the conf files (if they are modified).
*
* @param core the SolrCore
- * @param force force a replication in all cases
+ * @param forceReplication force a replication in all cases
* @return true on success, false if slave is already in sync
* @throws IOException if an exception occurs
*/
- @SuppressWarnings("unchecked")
- boolean fetchLatestIndex(SolrCore core, boolean force) throws IOException, InterruptedException {
+ boolean fetchLatestIndex(SolrCore core, boolean forceReplication) throws IOException, InterruptedException {
successfulInstall = false;
replicationStartTime = System.currentTimeMillis();
try {
@@ -278,7 +277,7 @@ public class SnapPuller {
}
if (latestVersion == 0L) {
- if (force && commit.getGeneration() != 0) {
+ if (forceReplication && commit.getGeneration() != 0) {
// since we won't get the files for an empty index,
// we just clear ours and commit
RefCounted<IndexWriter> iw = core.getUpdateHandler().getSolrCoreState().getIndexWriter(core);
@@ -297,7 +296,7 @@ public class SnapPuller {
return true;
}
- if (!force && IndexDeletionPolicyWrapper.getCommitTimestamp(commit) == latestVersion) {
+ if (!forceReplication && IndexDeletionPolicyWrapper.getCommitTimestamp(commit) == latestVersion) {
//master and slave are already in sync just return
LOG.info("Slave in sync with master.");
successfulInstall = true;
@@ -318,10 +317,11 @@ public class SnapPuller {
filesDownloaded = Collections.synchronizedList(new ArrayList<Map<String, Object>>());
// if the generateion of master is older than that of the slave , it means they are not compatible to be copied
// then a new index direcory to be created and all the files need to be copied
- boolean isFullCopyNeeded = IndexDeletionPolicyWrapper.getCommitTimestamp(commit) >= latestVersion || force;
+ boolean isFullCopyNeeded = IndexDeletionPolicyWrapper.getCommitTimestamp(commit) >= latestVersion || forceReplication;
File tmpIndexDir = createTempindexDir(core);
- if (isIndexStale())
+ if (isIndexStale()) {
isFullCopyNeeded = true;
+ }
LOG.info("Starting download to " + tmpIndexDir + " fullCopy=" + isFullCopyNeeded);
successfulInstall = false;
boolean deleteTmpIdxDir = true;
@@ -384,7 +384,7 @@ public class SnapPuller {
// may be closed
core.getDirectoryFactory().doneWithDirectory(oldDirectory);
}
- doCommit();
+ doCommit(isFullCopyNeeded);
}
replicationStartTime = 0;
@@ -533,11 +533,11 @@ public class SnapPuller {
return sb;
}
- private void doCommit() throws IOException {
+ private void doCommit(boolean isFullCopyNeeded) throws IOException {
SolrQueryRequest req = new LocalSolrQueryRequest(solrCore,
new ModifiableSolrParams());
// reboot the writer on the new index and get a new searcher
- solrCore.getUpdateHandler().newIndexWriter(true);
+ solrCore.getUpdateHandler().newIndexWriter(isFullCopyNeeded);
try {
// first try to open an NRT searcher so that the new
@@ -555,7 +555,10 @@ public class SnapPuller {
}
// update our commit point to the right dir
- solrCore.getUpdateHandler().commit(new CommitUpdateCommand(req, false));
+ CommitUpdateCommand cuc = new CommitUpdateCommand(req, false);
+ cuc.waitSearcher = false;
+ cuc.openSearcher = false;
+ solrCore.getUpdateHandler().commit(cuc);
} finally {
req.close();
@@ -979,6 +982,14 @@ public class SnapPuller {
this.file = new File(copy2Dir, saveAs);
+ File parentDir = this.file.getParentFile();
+ if( ! parentDir.exists() ){
+ if ( ! parentDir.mkdirs() ) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
+ "Failed to create (sub)directory for file: " + saveAs);
+ }
+ }
+
this.fileOutputStream = new FileOutputStream(file);
this.fileChannel = this.fileOutputStream.getChannel();
@@ -1151,7 +1162,7 @@ public class SnapPuller {
params.set(FILE, fileName);
}
if (useInternal) {
- params.set(COMPRESSION, "internal");
+ params.set(COMPRESSION, "true");
}
//use checksum
if (this.includeChecksum) {
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/UpdateRequestHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/UpdateRequestHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/UpdateRequestHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/UpdateRequestHandler.java Fri Sep 21 17:21:34 2012
@@ -74,7 +74,7 @@ public class UpdateRequestHandler extend
if(type == null) {
type = stream.getContentType();
}
- if( type == null ) { // Normal requests will not get here.
+ if( type == null ) { // Normal requests will not get here.
throw new SolrException(ErrorCode.BAD_REQUEST, "Missing ContentType");
}
int idx = type.indexOf(';');
Modified: lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java?rev=1388574&r1=1388573&r2=1388574&view=diff
==============================================================================
--- lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java (original)
+++ lucene/dev/branches/LUCENE-2878/solr/core/src/java/org/apache/solr/handler/admin/CollectionsHandler.java Fri Sep 21 17:21:34 2012
@@ -176,13 +176,13 @@ public class CollectionsHandler extends
private void handleCreateAction(SolrQueryRequest req,
SolrQueryResponse rsp) throws InterruptedException, KeeperException {
log.info("Creating Collection : " + req.getParamString());
- Integer numReplicas = req.getParams().getInt("numReplicas", 0);
+ Integer numReplicas = req.getParams().getInt(OverseerCollectionProcessor.REPLICATION_FACTOR, 0);
String name = req.getParams().required().get("name");
String configName = req.getParams().get("collection.configName");
String numShards = req.getParams().get("numShards");
ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION,
- OverseerCollectionProcessor.CREATECOLLECTION, "numReplicas", numReplicas.toString(), "name", name,
+ OverseerCollectionProcessor.CREATECOLLECTION, OverseerCollectionProcessor.REPLICATION_FACTOR, numReplicas.toString(), "name", name,
"collection.configName", configName, "numShards", numShards);
// TODO: what if you want to block until the collection is available?