mirror of https://github.com/apache/lucene.git
SOLR-5609 Reverting accidental commit
git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1567014 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Noble Paul
parent
1588fd5f5a
commit
ce3b4becb4
|
|
@ -40,8 +40,6 @@ import org.apache.solr.common.cloud.SolrZkClient;
|
|||
import org.apache.solr.common.cloud.ZkCoreNodeProps;
|
||||
import org.apache.solr.common.cloud.ZkNodeProps;
|
||||
import org.apache.solr.common.cloud.ZkStateReader;
|
||||
import org.apache.solr.common.params.CollectionParams;
|
||||
import org.apache.solr.common.util.NamedList;
|
||||
import org.apache.solr.handler.component.ShardHandler;
|
||||
import org.apache.zookeeper.CreateMode;
|
||||
import org.apache.zookeeper.KeeperException;
|
||||
|
|
@ -49,8 +47,6 @@ import org.slf4j.Logger;
|
|||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import static java.util.Collections.singletonMap;
|
||||
import static org.apache.solr.common.cloud.ZkNodeProps.makeMap;
|
||||
import static org.apache.solr.common.params.CollectionParams.CollectionAction.CLUSTERPROP;
|
||||
|
||||
/**
|
||||
* Cluster leader. Responsible node assignments, cluster state file?
|
||||
|
|
@ -62,7 +58,6 @@ public class Overseer {
|
|||
public static final String REMOVESHARD = "removeshard";
|
||||
public static final String ADD_ROUTING_RULE = "addroutingrule";
|
||||
public static final String REMOVE_ROUTING_RULE = "removeroutingrule";
|
||||
public static final String ADDREPLICA = "addreplica";
|
||||
|
||||
public static final int STATE_UPDATE_DELAY = 1500; // delay between cloud state updates
|
||||
|
||||
|
|
@ -71,7 +66,7 @@ public class Overseer {
|
|||
static enum LeaderStatus { DONT_KNOW, NO, YES };
|
||||
|
||||
private long lastUpdatedTime = 0;
|
||||
|
||||
|
||||
private class ClusterStateUpdater implements Runnable, ClosableThread {
|
||||
|
||||
private final ZkStateReader reader;
|
||||
|
|
@ -83,15 +78,13 @@ public class Overseer {
|
|||
//If Overseer dies while extracting the main queue a new overseer will start from this queue
|
||||
private final DistributedQueue workQueue;
|
||||
private volatile boolean isClosed;
|
||||
private Map clusterProps;
|
||||
|
||||
|
||||
public ClusterStateUpdater(final ZkStateReader reader, final String myId) {
|
||||
this.zkClient = reader.getZkClient();
|
||||
this.stateUpdateQueue = getInQueue(zkClient);
|
||||
this.workQueue = getInternalQueue(zkClient);
|
||||
this.myId = myId;
|
||||
this.reader = reader;
|
||||
clusterProps = reader.getClusterProps();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -228,12 +221,7 @@ public class Overseer {
|
|||
private ClusterState processMessage(ClusterState clusterState,
|
||||
final ZkNodeProps message, final String operation) {
|
||||
if ("state".equals(operation)) {
|
||||
if( isLegacy(message.getStr("collection"))) {
|
||||
clusterState = updateState(clusterState, message);
|
||||
} else {
|
||||
clusterState = updateStateNew(clusterState, message);
|
||||
}
|
||||
|
||||
clusterState = updateState(clusterState, message);
|
||||
} else if (DELETECORE.equals(operation)) {
|
||||
clusterState = removeCore(clusterState, message);
|
||||
} else if (REMOVECOLLECTION.equals(operation)) {
|
||||
|
|
@ -260,60 +248,16 @@ public class Overseer {
|
|||
clusterState = updateShardState(clusterState, message);
|
||||
} else if (OverseerCollectionProcessor.CREATECOLLECTION.equals(operation)) {
|
||||
clusterState = buildCollection(clusterState, message);
|
||||
} else if(ADDREPLICA.equals(operation)){
|
||||
clusterState = addReplica(clusterState,message);
|
||||
} else if (Overseer.ADD_ROUTING_RULE.equals(operation)) {
|
||||
clusterState = addRoutingRule(clusterState, message);
|
||||
} else if (Overseer.REMOVE_ROUTING_RULE.equals(operation)) {
|
||||
clusterState = removeRoutingRule(clusterState, message);
|
||||
} else if(CLUSTERPROP.isEqual(operation)){
|
||||
handleProp(message);
|
||||
|
||||
} else {
|
||||
throw new RuntimeException("unknown operation:" + operation
|
||||
+ " contents:" + message.getProperties());
|
||||
}
|
||||
return clusterState;
|
||||
}
|
||||
private void handleProp(ZkNodeProps message) {
|
||||
String name = message.getStr("name");
|
||||
String val = message.getStr("val");
|
||||
Map m = reader.getClusterProps();
|
||||
if(val ==null) m.remove(name);
|
||||
else m.put(name,val);
|
||||
|
||||
try {
|
||||
if(reader.getZkClient().exists(ZkStateReader.CLUSTER_PROPS,true))
|
||||
reader.getZkClient().setData(ZkStateReader.CLUSTER_PROPS,ZkStateReader.toJSON(m),true);
|
||||
else
|
||||
reader.getZkClient().create(ZkStateReader.CLUSTER_PROPS, ZkStateReader.toJSON(m),CreateMode.PERSISTENT, true);
|
||||
clusterProps = reader.getClusterProps();
|
||||
} catch (Exception e) {
|
||||
log.error("Unable to set cluster property", e);
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private ClusterState addReplica(ClusterState clusterState, ZkNodeProps message) {
|
||||
log.info("addReplica() {} ", message);
|
||||
String coll = message.getStr(ZkStateReader.COLLECTION_PROP);
|
||||
String slice = message.getStr(ZkStateReader.SHARD_ID_PROP);
|
||||
Slice sl = clusterState.getSlice(coll, slice);
|
||||
if(sl == null){
|
||||
log.error("Invalid Collection/Slice {}/{} ",coll,slice);
|
||||
return clusterState;
|
||||
}
|
||||
|
||||
String coreNodeName = Assign.assignNode(coll, clusterState);
|
||||
Replica replica = new Replica(coreNodeName,
|
||||
makeMap(
|
||||
ZkStateReader.CORE_NAME_PROP, message.getStr(ZkStateReader.CORE_NAME_PROP),
|
||||
ZkStateReader.BASE_URL_PROP,message.getStr(ZkStateReader.BASE_URL_PROP),
|
||||
ZkStateReader.STATE_PROP,message.getStr(ZkStateReader.STATE_PROP),
|
||||
ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName));
|
||||
sl.getReplicasMap().put(coreNodeName, replica);
|
||||
return clusterState;
|
||||
}
|
||||
|
||||
private ClusterState buildCollection(ClusterState clusterState, ZkNodeProps message) {
|
||||
String collection = message.getStr("name");
|
||||
|
|
@ -467,23 +411,6 @@ public class Overseer {
|
|||
log.info("According to ZK I (id=" + myId + ") am no longer a leader.");
|
||||
return LeaderStatus.NO;
|
||||
}
|
||||
|
||||
private ClusterState updateStateNew(ClusterState clusterState, ZkNodeProps message) {
|
||||
String collection = message.getStr(ZkStateReader.COLLECTION_PROP);
|
||||
String sliceName = message.getStr(ZkStateReader.SHARD_ID_PROP);
|
||||
|
||||
if(collection==null || sliceName == null){
|
||||
log.error("Invalid collection and slice {}", message);
|
||||
return clusterState;
|
||||
}
|
||||
Slice slice = clusterState.getSlice(collection, sliceName);
|
||||
if(slice == null){
|
||||
log.error("No such slice exists {}", message);
|
||||
return clusterState;
|
||||
}
|
||||
|
||||
return updateState(clusterState, message);
|
||||
}
|
||||
|
||||
/**
|
||||
* Try to assign core to the cluster.
|
||||
|
|
@ -604,16 +531,6 @@ public class Overseer {
|
|||
return newClusterState;
|
||||
}
|
||||
|
||||
private boolean isLegacy(String collection) {
|
||||
if("false".equals(clusterProps.get(OverseerCollectionProcessor.LEGACY_CLOUD)) ){
|
||||
return false;
|
||||
} else {
|
||||
return "defaultcol".equals(collection) || "collection1".equals(collection);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private ClusterState checkAndCompleteShardSplit(ClusterState state, String collection, String coreNodeName, String sliceName, Map<String,Object> replicaProps) {
|
||||
Slice slice = state.getSlice(collection, sliceName);
|
||||
Map<String, Object> sliceProps = slice.getProperties();
|
||||
|
|
@ -776,7 +693,7 @@ public class Overseer {
|
|||
// without explicitly creating a collection. In this current case, we assume custom sharding with an "implicit" router.
|
||||
slices = new HashMap<String, Slice>(1);
|
||||
props = new HashMap<String,Object>(1);
|
||||
props.put(DocCollection.DOC_ROUTER, makeMap("name", ImplicitDocRouter.NAME));
|
||||
props.put(DocCollection.DOC_ROUTER, ZkNodeProps.makeMap("name",ImplicitDocRouter.NAME));
|
||||
router = new ImplicitDocRouter();
|
||||
} else {
|
||||
props = coll.getProperties();
|
||||
|
|
@ -1045,8 +962,7 @@ public class Overseer {
|
|||
private ShardHandler shardHandler;
|
||||
|
||||
private String adminPath;
|
||||
|
||||
private OverseerCollectionProcessor ocp;
|
||||
|
||||
// overseer not responsible for closing reader
|
||||
public Overseer(ShardHandler shardHandler, String adminPath, final ZkStateReader reader) throws KeeperException, InterruptedException {
|
||||
this.reader = reader;
|
||||
|
|
@ -1064,9 +980,7 @@ public class Overseer {
|
|||
updaterThread.setDaemon(true);
|
||||
|
||||
ThreadGroup ccTg = new ThreadGroup("Overseer collection creation process.");
|
||||
|
||||
ocp = new OverseerCollectionProcessor(reader, id, shardHandler, adminPath);
|
||||
ccThread = new OverseerThread(ccTg, ocp,
|
||||
ccThread = new OverseerThread(ccTg, new OverseerCollectionProcessor(reader, id, shardHandler, adminPath),
|
||||
"Overseer-" + id);
|
||||
ccThread.setDaemon(true);
|
||||
|
||||
|
|
|
|||
|
|
@ -126,15 +126,13 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
|
||||
public static final String COLL_PROP_PREFIX = "property.";
|
||||
|
||||
public static final String LEGACY_CLOUD ="legacyCloud";
|
||||
|
||||
public static final Set<String> KNOWN_CLUSTER_PROPS = ImmutableSet.of(LEGACY_CLOUD);
|
||||
public static final Set<String> KNOWN_CLUSTER_PROPS = ImmutableSet.of("legacyCloud");
|
||||
|
||||
public static final Map<String,Object> COLL_PROPS = ZkNodeProps.makeMap(
|
||||
ROUTER, DocRouter.DEFAULT_NAME,
|
||||
REPLICATION_FACTOR, "1",
|
||||
MAX_SHARDS_PER_NODE, "1",
|
||||
"external", null);
|
||||
"external",null );
|
||||
|
||||
|
||||
// TODO: use from Overseer?
|
||||
|
|
@ -513,7 +511,6 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
|
||||
ShardRequest sreq = new ShardRequest();
|
||||
sreq.purpose = 1;
|
||||
if (baseUrl.startsWith("http://")) baseUrl = baseUrl.substring(7);
|
||||
sreq.shards = new String[] {baseUrl};
|
||||
sreq.actualShards = sreq.shards;
|
||||
sreq.params = new ModifiableSolrParams(new MapSolrParams(m));
|
||||
|
|
@ -832,7 +829,13 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
}
|
||||
|
||||
// find the leader for the shard
|
||||
Replica parentShardLeader = clusterState.getLeader(collectionName, slice);
|
||||
Replica parentShardLeader = null;
|
||||
try {
|
||||
parentShardLeader = zkStateReader.getLeaderRetry(collectionName, slice, 10000);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
DocRouter.Range range = parentSlice.getRange();
|
||||
if (range == null) {
|
||||
range = new PlainIdRouter().fullRange();
|
||||
|
|
@ -1347,7 +1350,7 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
}
|
||||
log.info("Common hash range between source shard: {} and target shard: {} = " + splitRange, sourceSlice.getName(), targetSlice.getName());
|
||||
|
||||
Replica targetLeader = targetSlice.getLeader();
|
||||
Replica targetLeader = zkStateReader.getLeaderRetry(targetCollection.getName(), targetSlice.getName(), 10000);
|
||||
|
||||
log.info("Asking target leader node: " + targetLeader.getNodeName() + " core: "
|
||||
+ targetLeader.getStr("core") + " to buffer updates");
|
||||
|
|
@ -1391,7 +1394,7 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
log.info("Routing rule added successfully");
|
||||
|
||||
// Create temp core on source shard
|
||||
Replica sourceLeader = sourceSlice.getLeader();
|
||||
Replica sourceLeader = zkStateReader.getLeaderRetry(sourceCollection.getName(), sourceSlice.getName(), 10000);
|
||||
|
||||
// create a temporary collection with just one node on the shard leader
|
||||
String configName = zkStateReader.readConfigName(sourceCollection.getName());
|
||||
|
|
@ -1407,7 +1410,7 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
// refresh cluster state
|
||||
clusterState = zkStateReader.getClusterState();
|
||||
Slice tempSourceSlice = clusterState.getCollection(tempSourceCollectionName).getSlices().iterator().next();
|
||||
Replica tempSourceLeader = zkStateReader.getLeaderRetry(tempSourceCollectionName, tempSourceSlice.getName(), 60000);
|
||||
Replica tempSourceLeader = zkStateReader.getLeaderRetry(tempSourceCollectionName, tempSourceSlice.getName(), 120000);
|
||||
|
||||
String tempCollectionReplica1 = tempSourceCollectionName + "_" + tempSourceSlice.getName() + "_replica1";
|
||||
String coreNodeName = waitForCoreNodeName(clusterState.getCollection(tempSourceCollectionName),
|
||||
|
|
@ -1744,8 +1747,6 @@ public class OverseerCollectionProcessor implements Runnable, ClosableThread {
|
|||
// yes, they must use same admin handler path everywhere...
|
||||
cloneParams.set("qt", adminPath);
|
||||
sreq.purpose = 1;
|
||||
// TODO: this sucks
|
||||
if (replica.startsWith("http://")) replica = replica.substring(7);
|
||||
sreq.shards = new String[] {replica};
|
||||
sreq.actualShards = sreq.shards;
|
||||
sreq.params = cloneParams;
|
||||
|
|
|
|||
|
|
@ -71,7 +71,6 @@ import static org.apache.solr.common.cloud.ZkNodeProps.makeMap;
|
|||
import static org.apache.solr.common.cloud.ZkStateReader.COLLECTION_PROP;
|
||||
import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP;
|
||||
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDROLE;
|
||||
import static org.apache.solr.common.params.CollectionParams.CollectionAction.CLUSTERPROP;
|
||||
import static org.apache.solr.common.params.CollectionParams.CollectionAction.REMOVEROLE;
|
||||
|
||||
public class CollectionsHandler extends RequestHandlerBase {
|
||||
|
|
@ -208,11 +207,11 @@ public class CollectionsHandler extends RequestHandlerBase {
|
|||
}
|
||||
|
||||
Map<String,Object> props = ZkNodeProps.makeMap(
|
||||
Overseer.QUEUE_OPERATION, CLUSTERPROP.toLower() );
|
||||
Overseer.QUEUE_OPERATION, CollectionAction.CLUSTERPROP.toString().toLowerCase(Locale.ROOT) );
|
||||
copyIfNotNull(req.getParams(),props,
|
||||
"name",
|
||||
"val");
|
||||
handleResponse(CLUSTERPROP.toLower(),new ZkNodeProps(props),rsp);
|
||||
handleResponse(CollectionAction.CLUSTERPROP.toString().toLowerCase(Locale.ROOT),new ZkNodeProps(props),rsp);
|
||||
|
||||
}
|
||||
|
||||
|
|
@ -220,11 +219,11 @@ public class CollectionsHandler extends RequestHandlerBase {
|
|||
|
||||
private void handleRole(CollectionAction action, SolrQueryRequest req, SolrQueryResponse rsp) throws KeeperException, InterruptedException {
|
||||
req.getParams().required().check("role", "node");
|
||||
Map<String, Object> map = ZkNodeProps.makeMap(Overseer.QUEUE_OPERATION, action.toLower());
|
||||
Map<String, Object> map = ZkNodeProps.makeMap(Overseer.QUEUE_OPERATION, action.toString().toLowerCase(Locale.ROOT));
|
||||
copyIfNotNull(req.getParams(), map,"role", "node");
|
||||
ZkNodeProps m = new ZkNodeProps(map);
|
||||
if(!KNOWN_ROLES.contains(m.getStr("role"))) throw new SolrException(ErrorCode.BAD_REQUEST,"Unknown role. Supported roles are ,"+ KNOWN_ROLES);
|
||||
Overseer.getInQueue(coreContainer.getZkController().getZkClient()).offer(ZkStateReader.toJSON(m)) ;
|
||||
handleResponse(action.toString().toLowerCase(Locale.ROOT), m, rsp);
|
||||
}
|
||||
|
||||
public static long DEFAULT_ZK_TIMEOUT = 180*1000;
|
||||
|
|
|
|||
|
|
@ -197,16 +197,16 @@ public class CollectionsAPIDistributedZkTest extends AbstractFullDistribZkTestBa
|
|||
@Override
|
||||
public void doTest() throws Exception {
|
||||
testSolrJAPICalls();
|
||||
// testNodesUsedByCreate();
|
||||
// testCollectionsAPI();
|
||||
// testCollectionsAPIAddRemoveStress();
|
||||
// testErrorHandling();
|
||||
// deletePartiallyCreatedCollection();
|
||||
// deleteCollectionRemovesStaleZkCollectionsNode();
|
||||
// clusterPropTest();
|
||||
//
|
||||
// last
|
||||
// deleteCollectionWithDownNodes();
|
||||
testNodesUsedByCreate();
|
||||
testCollectionsAPI();
|
||||
testCollectionsAPIAddRemoveStress();
|
||||
testErrorHandling();
|
||||
deletePartiallyCreatedCollection();
|
||||
deleteCollectionRemovesStaleZkCollectionsNode();
|
||||
clusterPropTest();
|
||||
|
||||
// last
|
||||
deleteCollectionWithDownNodes();
|
||||
if (DEBUG) {
|
||||
super.printLayout();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -53,13 +53,5 @@ public interface CollectionParams
|
|||
}
|
||||
return null;
|
||||
}
|
||||
public boolean isEqual(String s){
|
||||
if(s == null) return false;
|
||||
return toString().equals(s.toUpperCase(Locale.ROOT));
|
||||
}
|
||||
public String toLower(){
|
||||
return toString().toLowerCase(Locale.ROOT);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue