mirror of
https://github.com/yacy/yacy_search_server.git
synced 2024-09-19 00:01:41 +02:00
added SolrCloud access mode and configuration
This commit is contained in:
Michael Peter Christen
parent
8514bffc22
commit
d07cdd8c3b
|
|
@ -64,7 +64,12 @@
|
|||
When a search request is made, all servers are accessed synchronously and the result is combined.</dd>
|
||||
|
||||
<dt class="TableCellDark"></dt>
|
||||
<dd>Sharding Method<br/><input type="text" size="50" maxlength="50" value="#[solr.indexing.sharding]#" name="solr.indexing.sharding" id="solr.indexing.sharding" disabled="disabled"/></dd>
|
||||
<dd>Sharding Method<br/>
|
||||
<select id="solr.indexing.sharding" name="solr.indexing.sharding">
|
||||
#{solr.indexing.sharding.methods}#
|
||||
<option value="#[method]#" #(selected)#::selected="selected"#(/selected)#>#[method]#: #[description]#</option>
|
||||
#{/solr.indexing.sharding.methods}#
|
||||
</select></dd>
|
||||
|
||||
<dt class="TableCellDark"></dt>
|
||||
<dd><input type="checkbox" name="solr.indexing.solrremote.writeenabled" id="solr_indexing_solrremote_writeenabled" #(solr.indexing.solrremote.writeenabled.checked)#:: checked="checked"#(/solr.indexing.solrremote.writeenabled.checked)#/> write-enabled (if unchecked, the remote server(s) will only be used as search peers)</dd>
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ import org.apache.solr.common.SolrException;
|
|||
|
||||
import net.yacy.cora.document.encoding.UTF8;
|
||||
import net.yacy.cora.federate.solr.connector.RemoteSolrConnector;
|
||||
import net.yacy.cora.federate.solr.connector.ShardSelection;
|
||||
import net.yacy.cora.federate.solr.connector.SolrConnector;
|
||||
import net.yacy.cora.federate.solr.instance.RemoteInstance;
|
||||
import net.yacy.cora.federate.solr.instance.ShardInstance;
|
||||
|
|
@ -110,7 +111,9 @@ public class IndexFederated_p {
|
|||
}
|
||||
solrurls = s.toString().trim();
|
||||
env.setConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_URL, solrurls);
|
||||
env.setConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, post.get("solr.indexing.sharding", env.getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, "modulo-host-md5")));
|
||||
String shardMethodName = post.get("solr.indexing.sharding", env.getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, ShardSelection.Method.MODULO_HOST_MD5.name()));
|
||||
ShardSelection.Method shardMethod = ShardSelection.Method.valueOf(shardMethodName);
|
||||
env.setConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, shardMethod.name());
|
||||
|
||||
if (solrRemoteWasOn && !solrRemoteIsOnAfterwards) {
|
||||
// switch off
|
||||
|
|
@ -133,7 +136,7 @@ public class IndexFederated_p {
|
|||
try {
|
||||
if (usesolr) {
|
||||
ArrayList<RemoteInstance> instances = RemoteInstance.getShardInstances(solrurls, null, null, solrtimeout);
|
||||
sb.index.fulltext().connectRemoteSolr(instances, writeEnabled);
|
||||
sb.index.fulltext().connectRemoteSolr(instances, shardMethod, writeEnabled);
|
||||
} else {
|
||||
sb.index.fulltext().disconnectRemoteSolr();
|
||||
}
|
||||
|
|
@ -176,7 +179,15 @@ public class IndexFederated_p {
|
|||
prop.put(SwitchboardConstants.CORE_SERVICE_WEBGRAPH + ".checked", env.getConfigBool(SwitchboardConstants.CORE_SERVICE_WEBGRAPH, false) ? 1 : 0);
|
||||
prop.put("solr.indexing.solrremote.checked", env.getConfigBool(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_ENABLED, false) ? 1 : 0);
|
||||
prop.put("solr.indexing.url", env.getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_URL, "http://127.0.0.1:8983/solr").replace(",", "\n"));
|
||||
prop.put("solr.indexing.sharding", env.getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, "modulo-host-md5"));
|
||||
String thisShardingMethodName = env.getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, ShardSelection.Method.MODULO_HOST_MD5.name());
|
||||
int mc = 0;
|
||||
for (ShardSelection.Method method: ShardSelection.Method.values()) {
|
||||
prop.put("solr.indexing.sharding.methods_" + mc + "_method", method.name());
|
||||
prop.put("solr.indexing.sharding.methods_" + mc + "_description", method.description);
|
||||
prop.put("solr.indexing.sharding.methods_" + mc + "_selected", method.name().equals(thisShardingMethodName) ? 1 : 0);
|
||||
mc++;
|
||||
}
|
||||
prop.put("solr.indexing.sharding.methods", mc);
|
||||
prop.put("solr.indexing.solrremote.writeenabled.checked", env.getConfigBool(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_WRITEENABLED, true));
|
||||
prop.put("solr.indexing.lazy.checked", env.getConfigBool(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_LAZY, true) ? 1 : 0);
|
||||
|
||||
|
|
|
|||
|
|
@ -21,91 +21,115 @@
|
|||
package net.yacy.cora.federate.solr.connector;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.MalformedURLException;
|
||||
import java.net.URL;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import net.yacy.cora.document.encoding.ASCII;
|
||||
import net.yacy.cora.util.ConcurrentLog;
|
||||
import net.yacy.search.schema.CollectionSchema;
|
||||
|
||||
import org.apache.solr.client.solrj.SolrServer;
|
||||
import org.apache.solr.common.SolrInputDocument;
|
||||
import org.apache.solr.common.SolrInputField;
|
||||
|
||||
public class ShardSelection {
|
||||
public class ShardSelection implements Iterable<SolrServer> {
|
||||
|
||||
private final Method method; // the sharding method
|
||||
private final AtomicLong chardID; // the next id that shall be given away
|
||||
private final AtomicLong shardID; // the next id that shall be given away
|
||||
private final int dimension; // the number of chards
|
||||
private final ArrayList<SolrServer> server;
|
||||
|
||||
public enum Method {
|
||||
MODULO_HOST_MD5, ROUND_ROBIN;
|
||||
MODULO_HOST_MD5("hash-based calculation of storage targets, select all for retrieval"),
|
||||
ROUND_ROBIN("round-robin of storage targets, select all for retrieval"),
|
||||
SOLRCLOUD("round-robin of storage targets and round-robin for retrieval");
|
||||
public final String description;
|
||||
private Method(final String description) {
|
||||
this.description = description;
|
||||
}
|
||||
}
|
||||
|
||||
public ShardSelection(final Method method, final int dimension) {
|
||||
public ShardSelection(final ArrayList<SolrServer> server, final Method method) {
|
||||
this.server = server;
|
||||
this.method = method;
|
||||
this.dimension = dimension;
|
||||
this.chardID = new AtomicLong(0);
|
||||
this.dimension = server.size();
|
||||
this.shardID = new AtomicLong(0);
|
||||
}
|
||||
|
||||
public Method getMethod() {
|
||||
return this.method;
|
||||
}
|
||||
|
||||
private int selectRoundRobin() {
|
||||
return (int) (this.chardID.getAndIncrement() % this.dimension);
|
||||
int rr = (int) (this.shardID.getAndIncrement() % this.dimension);
|
||||
if (this.shardID.get() < 0) this.shardID.set(0);
|
||||
return rr;
|
||||
}
|
||||
|
||||
public int select(final SolrInputDocument solrdoc) throws IOException {
|
||||
public SolrServer server4write(final SolrInputDocument solrdoc) throws IOException {
|
||||
if (this.method == Method.MODULO_HOST_MD5) {
|
||||
SolrInputField sif = solrdoc.getField(CollectionSchema.host_s.getSolrFieldName());
|
||||
if (sif != null) {
|
||||
final String host = (String) sif.getValue();
|
||||
return selectHost(host);
|
||||
if (host != null && host.length() > 0) return server4write(host);
|
||||
}
|
||||
sif = solrdoc.getField(CollectionSchema.sku.getSolrFieldName());
|
||||
if (sif != null) {
|
||||
final String url = (String) sif.getValue();
|
||||
try {
|
||||
return selectURL(url);
|
||||
if (url != null && url.length() > 0) try {
|
||||
return server4write(new URL(url));
|
||||
} catch (final IOException e) {
|
||||
ConcurrentLog.logException(e);
|
||||
return 0;
|
||||
return this.server.get(0);
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
return this.server.get(0);
|
||||
}
|
||||
|
||||
// finally if no method matches use ROUND_ROBIN
|
||||
return selectRoundRobin();
|
||||
return this.server.get(selectRoundRobin());
|
||||
}
|
||||
|
||||
public int selectHost(final String host) throws IOException {
|
||||
if (this.method == Method.MODULO_HOST_MD5) {
|
||||
public SolrServer server4write(final String host) throws IOException {
|
||||
if (host == null) throw new IOException("sharding - host url, host empty: " + host);
|
||||
if (host.indexOf("://") >= 0) return server4write(new URL(host)); // security catch for accidantly using the wrong method
|
||||
if (this.method == Method.MODULO_HOST_MD5) {
|
||||
try {
|
||||
final MessageDigest digest = MessageDigest.getInstance("MD5");
|
||||
digest.update(ASCII.getBytes(host));
|
||||
final byte[] md5 = digest.digest();
|
||||
return (0xff & md5[0]) % this.dimension;
|
||||
return this.server.get((0xff & md5[0]) % this.dimension);
|
||||
} catch (final NoSuchAlgorithmException e) {
|
||||
throw new IOException("sharding - no md5 available: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// finally if no method matches use ROUND_ROBIN
|
||||
return (int) (this.chardID.getAndIncrement() % this.dimension);
|
||||
return this.server.get(selectRoundRobin());
|
||||
}
|
||||
|
||||
public int selectURL(final String sku) throws IOException {
|
||||
if (this.method == Method.MODULO_HOST_MD5) {
|
||||
try {
|
||||
final URL url = new URL(sku);
|
||||
final String host = url.getHost();
|
||||
return selectHost(host);
|
||||
} catch (final MalformedURLException e) {
|
||||
throw new IOException("sharding - bad url: " + sku);
|
||||
}
|
||||
public SolrServer server4write(final URL url) throws IOException {
|
||||
return server4write(url.getHost());
|
||||
}
|
||||
|
||||
// finally if no method matches use ROUND_ROBIN
|
||||
return (int) (this.chardID.getAndIncrement() % this.dimension);
|
||||
public List<SolrServer> server4read() {
|
||||
if (this.method == Method.MODULO_HOST_MD5 || this.method == Method.ROUND_ROBIN) return this.server; // return all
|
||||
// this is a SolrCloud, we select just one of the SolrCloud server(s)
|
||||
ArrayList<SolrServer> a = new ArrayList<>(1);
|
||||
a.add(this.server.get(selectRoundRobin()));
|
||||
return a;
|
||||
}
|
||||
|
||||
/**
|
||||
* return all solr server
|
||||
*/
|
||||
@Override
|
||||
public Iterator<SolrServer> iterator() {
|
||||
return this.server.iterator();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,13 +51,11 @@ public class ServerShard extends SolrServer {
|
|||
_dummyOKResponse.setResponse(new NamedList<Object>());
|
||||
}
|
||||
|
||||
private final ArrayList<SolrServer> server;
|
||||
private final ShardSelection sharding;
|
||||
private final ShardSelection shards;
|
||||
private final boolean writeEnabled;
|
||||
|
||||
public ServerShard(ArrayList<SolrServer> server, final ShardSelection.Method method, final boolean writeEnabled) {
|
||||
this.server = server;
|
||||
this.sharding = new ShardSelection(method, this.server.size());
|
||||
this.shards = new ShardSelection(server, method);
|
||||
this.writeEnabled = writeEnabled;
|
||||
}
|
||||
|
||||
|
|
@ -70,7 +68,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse add(Collection<SolrInputDocument> docs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrInputDocument doc: docs) ur = server.get(this.sharding.select(doc)).add(doc);
|
||||
for (SolrInputDocument doc: docs) ur = this.shards.server4write(doc).add(doc);
|
||||
return ur; // TODO: this accumlation of update responses is wrong, but sufficient (because we do not evaluate it)
|
||||
}
|
||||
|
||||
|
|
@ -85,7 +83,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse add(Collection<SolrInputDocument> docs, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrInputDocument doc: docs) ur = server.get(this.sharding.select(doc)).add(doc, commitWithinMs);
|
||||
for (SolrInputDocument doc: docs) ur = this.shards.server4write(doc).add(doc, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -98,7 +96,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse addBeans(Collection<?> beans ) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.addBeans(beans);
|
||||
for (SolrServer s: this.shards) ur = s.addBeans(beans);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -113,7 +111,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse addBeans(Collection<?> beans, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.addBeans(beans, commitWithinMs);
|
||||
for (SolrServer s: this.shards) ur = s.addBeans(beans, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -125,7 +123,7 @@ public class ServerShard extends SolrServer {
|
|||
@Override
|
||||
public UpdateResponse add(SolrInputDocument doc) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
return server.get(this.sharding.select(doc)).add(doc);
|
||||
return this.shards.server4write(doc).add(doc);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -138,7 +136,7 @@ public class ServerShard extends SolrServer {
|
|||
@Override
|
||||
public UpdateResponse add(SolrInputDocument doc, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
return server.get(this.sharding.select(doc)).add(doc, commitWithinMs);
|
||||
return this.shards.server4write(doc).add(doc, commitWithinMs);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -150,7 +148,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse addBean(Object obj) throws IOException, SolrServerException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.addBean(obj);
|
||||
for (SolrServer s: this.shards) ur = s.addBean(obj);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -165,7 +163,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse addBean(Object obj, int commitWithinMs) throws IOException, SolrServerException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.addBean(obj, commitWithinMs);
|
||||
for (SolrServer s: this.shards) ur = s.addBean(obj, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -179,7 +177,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse commit() throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.commit();
|
||||
for (SolrServer s: this.shards) ur = s.commit();
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -195,7 +193,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse optimize() throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.optimize();
|
||||
for (SolrServer s: this.shards) ur = s.optimize();
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -209,7 +207,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse commit(boolean waitFlush, boolean waitSearcher) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.commit(waitFlush, waitSearcher);
|
||||
for (SolrServer s: this.shards) ur = s.commit(waitFlush, waitSearcher);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -224,7 +222,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse commit(boolean waitFlush, boolean waitSearcher, boolean softCommit) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.commit(waitFlush, waitSearcher, softCommit);
|
||||
for (SolrServer s: this.shards) ur = s.commit(waitFlush, waitSearcher, softCommit);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -240,7 +238,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse optimize(boolean waitFlush, boolean waitSearcher) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.optimize(waitFlush, waitSearcher);
|
||||
for (SolrServer s: this.shards) ur = s.optimize(waitFlush, waitSearcher);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -257,7 +255,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse optimize(boolean waitFlush, boolean waitSearcher, int maxSegments) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.optimize(waitFlush, waitSearcher, maxSegments);
|
||||
for (SolrServer s: this.shards) ur = s.optimize(waitFlush, waitSearcher, maxSegments);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -273,7 +271,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse rollback() throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.rollback();
|
||||
for (SolrServer s: this.shards) ur = s.rollback();
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -286,7 +284,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteById(String id) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteById(id);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteById(id);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -301,7 +299,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteById(String id, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteById(id, commitWithinMs);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteById(id, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -314,7 +312,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteById(List<String> ids) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteById(ids);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteById(ids);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -329,7 +327,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteById(List<String> ids, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteById(ids, commitWithinMs);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteById(ids, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -342,7 +340,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteByQuery(String query) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteByQuery(query);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteByQuery(query);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -357,7 +355,7 @@ public class ServerShard extends SolrServer {
|
|||
public UpdateResponse deleteByQuery(String query, int commitWithinMs) throws SolrServerException, IOException {
|
||||
if (!this.writeEnabled) return _dummyOKResponse;
|
||||
UpdateResponse ur = null;
|
||||
for (SolrServer s: server) ur = s.deleteByQuery(query, commitWithinMs);
|
||||
for (SolrServer s: this.shards.server4read()) ur = s.deleteByQuery(query, commitWithinMs);
|
||||
return ur;
|
||||
}
|
||||
|
||||
|
|
@ -367,7 +365,7 @@ public class ServerShard extends SolrServer {
|
|||
*/
|
||||
@Override
|
||||
public SolrPingResponse ping() throws SolrServerException, IOException {
|
||||
for (SolrServer s: server) {
|
||||
for (SolrServer s: this.shards) {
|
||||
SolrPingResponse spr = s.ping();
|
||||
if (spr != null) return spr;
|
||||
}
|
||||
|
|
@ -380,11 +378,15 @@ public class ServerShard extends SolrServer {
|
|||
*/
|
||||
@Override
|
||||
public QueryResponse query(final SolrParams params) throws SolrServerException {
|
||||
List<SolrServer> qs = this.shards.server4read();
|
||||
if (qs.size() == 1) {
|
||||
return qs.get(0).query(params);
|
||||
}
|
||||
|
||||
final Collection<QueryResponse> qrl = new ConcurrentLinkedQueue<QueryResponse>();
|
||||
// concurrently call all shards
|
||||
final Collection<QueryResponse> qrl = new ConcurrentLinkedQueue<QueryResponse>();
|
||||
List<Thread> t = new ArrayList<Thread>();
|
||||
for (final SolrServer s: server) {
|
||||
for (final SolrServer s: qs) {
|
||||
Thread t0 = new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
|
@ -414,11 +416,15 @@ public class ServerShard extends SolrServer {
|
|||
*/
|
||||
@Override
|
||||
public QueryResponse query(final SolrParams params, final METHOD method) throws SolrServerException {
|
||||
List<SolrServer> qs = this.shards.server4read();
|
||||
if (qs.size() == 1) {
|
||||
return qs.get(0).query(params, method);
|
||||
}
|
||||
|
||||
final Collection<QueryResponse> qrl = new ConcurrentLinkedQueue<QueryResponse>();
|
||||
// concurrently call all shards
|
||||
List<Thread> t = new ArrayList<Thread>();
|
||||
for (final SolrServer s: server) {
|
||||
for (final SolrServer s: qs) {
|
||||
Thread t0 = new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
|
@ -464,14 +470,14 @@ public class ServerShard extends SolrServer {
|
|||
@Override
|
||||
public NamedList<Object> request(final SolrRequest request) throws SolrServerException, IOException {
|
||||
ResponseAccumulator acc = new ResponseAccumulator();
|
||||
for (SolrServer s: server) acc.addResponse(s.request(request));
|
||||
for (SolrServer s: this.shards.server4read()) acc.addResponse(s.request(request));
|
||||
return acc.getAccumulatedResponse();
|
||||
}
|
||||
|
||||
@Override
|
||||
public DocumentObjectBinder getBinder() {
|
||||
DocumentObjectBinder db;
|
||||
for (SolrServer s: server) {
|
||||
for (SolrServer s: this.shards) {
|
||||
db = s.getBinder();
|
||||
if (db != null) return db;
|
||||
}
|
||||
|
|
@ -480,7 +486,7 @@ public class ServerShard extends SolrServer {
|
|||
|
||||
@Override
|
||||
public void shutdown() {
|
||||
for (SolrServer s: server) {
|
||||
for (SolrServer s: this.shards) {
|
||||
s.shutdown();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -98,6 +98,7 @@ import net.yacy.cora.document.id.MultiProtocolURL;
|
|||
import net.yacy.cora.federate.solr.FailCategory;
|
||||
import net.yacy.cora.federate.solr.Ranking;
|
||||
import net.yacy.cora.federate.solr.SchemaConfiguration;
|
||||
import net.yacy.cora.federate.solr.connector.ShardSelection;
|
||||
import net.yacy.cora.federate.solr.connector.SolrConnector.Metadata;
|
||||
import net.yacy.cora.federate.solr.instance.RemoteInstance;
|
||||
import net.yacy.cora.federate.yacy.CacheStrategy;
|
||||
|
|
@ -527,7 +528,9 @@ public final class Switchboard extends serverSwitch {
|
|||
if (usesolr && solrurls != null && solrurls.length() > 0) {
|
||||
try {
|
||||
ArrayList<RemoteInstance> instances = RemoteInstance.getShardInstances(solrurls, null, null, solrtimeout);
|
||||
this.index.fulltext().connectRemoteSolr(instances, writeEnabled);
|
||||
String shardMethodName = getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, ShardSelection.Method.MODULO_HOST_MD5.name());
|
||||
ShardSelection.Method shardMethod = ShardSelection.Method.valueOf(shardMethodName);
|
||||
this.index.fulltext().connectRemoteSolr(instances, shardMethod, writeEnabled);
|
||||
} catch (final IOException e ) {
|
||||
ConcurrentLog.logException(e);
|
||||
}
|
||||
|
|
@ -1371,7 +1374,9 @@ public final class Switchboard extends serverSwitch {
|
|||
if (usesolr && solrurls != null && solrurls.length() > 0) {
|
||||
try {
|
||||
ArrayList<RemoteInstance> instances = RemoteInstance.getShardInstances(solrurls, null, null, solrtimeout);
|
||||
this.index.fulltext().connectRemoteSolr(instances, writeEnabled);
|
||||
String shardMethodName = getConfig(SwitchboardConstants.FEDERATED_SERVICE_SOLR_INDEXING_SHARDING, ShardSelection.Method.MODULO_HOST_MD5.name());
|
||||
ShardSelection.Method shardMethod = ShardSelection.Method.valueOf(shardMethodName);
|
||||
this.index.fulltext().connectRemoteSolr(instances, shardMethod, writeEnabled);
|
||||
} catch (final IOException e ) {
|
||||
ConcurrentLog.logException(e);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -148,8 +148,8 @@ public final class Fulltext {
|
|||
return this.solrInstances.isConnectedRemote();
|
||||
}
|
||||
|
||||
public void connectRemoteSolr(final ArrayList<RemoteInstance> instances, final boolean writeEnabled) {
|
||||
this.solrInstances.connectRemote(new ShardInstance(instances, ShardSelection.Method.MODULO_HOST_MD5, writeEnabled));
|
||||
public void connectRemoteSolr(final ArrayList<RemoteInstance> instances, final ShardSelection.Method shardMethod, final boolean writeEnabled) {
|
||||
this.solrInstances.connectRemote(new ShardInstance(instances, shardMethod, writeEnabled));
|
||||
}
|
||||
|
||||
public void disconnectRemoteSolr() {
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user