oscerd commented on a change in pull request #5613:
URL: https://github.com/apache/camel/pull/5613#discussion_r643219143
##########
File path:
components/camel-solr/src/main/java/org/apache/camel/component/solr/SolrComponent.java
##########
@@ -86,38 +113,105 @@ public SolrComponent() {
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
- Endpoint endpoint = new SolrEndpoint(uri, this, remaining);
+ SolrConfiguration configuration = new SolrConfiguration(uri,
remaining);
+ Endpoint endpoint = new SolrEndpoint(uri, this, configuration);
setProperties(endpoint, parameters);
return endpoint;
}
+ public SolrClient getSolrClient(SolrProducer solrProducer,
SolrConfiguration solrConfiguration) throws Exception {
+ String signature = solrConfiguration.getSignature();
+ SolrClientReference solrClientReference;
+ if (!solrClientMap.containsKey(signature)) {
+ solrClientReference = new
SolrClientReference(solrConfiguration.initSolrClient());
+ solrClientMap.put(signature, solrClientReference);
+ // backward compatibility
+ addSolrClientToSolrServerReference(solrConfiguration,
solrClientReference.getSolrClient());
+ } else {
+ solrClientReference = solrClientMap.get(signature);
+ }
+ // register producer against solrClient (for later close of client)
+ solrClientReference.registerSolrProducer(solrProducer);
+ return solrClientReference.getSolrClient();
+ }
+
+ private void addSolrClientToSolrServerReference(SolrConfiguration
solrConfiguration, SolrClient solrClient) {
+ SolrEndpoint solrEndpoint = solrConfiguration.getSolrEndpoint();
+ if (solrEndpoint != null) {
+ SolrServerReference solrServerReference =
servers.get(solrEndpoint);
+ if (solrServerReference == null) {
+ solrServerReference = new SolrServerReference();
+ servers.put(solrEndpoint, solrServerReference);
+ }
+ if (solrClient instanceof CloudSolrClient) {
+ solrServerReference.setCloudSolrServer((CloudSolrClient)
solrClient);
+ }
+ if (solrClient instanceof ConcurrentUpdateSolrClient) {
+
solrServerReference.setUpdateSolrServer((ConcurrentUpdateSolrClient)
solrClient);
+ }
+ if (solrClient instanceof HttpSolrClient) {
+ solrServerReference.setSolrServer((HttpSolrClient) solrClient);
+ }
+ }
+ }
+
+ public void closeSolrClient(SolrProducer solrProducer) {
+ // close when generated for endpoint
+ List<String> signatureToRemoveList = new ArrayList<>();
+ for (Map.Entry<String, SolrClientReference> entry :
solrClientMap.entrySet()) {
+ SolrClientReference solrClientReference = entry.getValue();
+ if (solrClientReference.unRegisterSolrProducer(solrProducer) == 0)
{
+ signatureToRemoveList.add(entry.getKey());
+ }
+ }
+ removeFromSolrClientMap(signatureToRemoveList);
+ }
+
+ private void removeFromSolrClientMap(Collection<String>
signatureToRemoveList) {
+ for (String signature : signatureToRemoveList) {
+ SolrClientReference solrClientReference =
solrClientMap.get(signature);
+ solrClientMap.remove(signature);
+ try {
+ solrClientReference.getSolrClient().close();
+ } catch (IOException e) {
+ LOG.warn("Error shutting down solr client. This exception is
ignored.", e);
+ }
+ }
+ }
+
+ @Deprecated
public SolrServerReference getSolrServers(SolrEndpoint endpoint) {
return servers.get(endpoint);
}
+ @Deprecated
public void addSolrServers(SolrEndpoint endpoint, SolrServerReference
servers) {
this.servers.put(endpoint, servers);
}
@Override
protected void doShutdown() throws Exception {
+ removeFromSolrClientMap(solrClientMap.keySet());
for (SolrServerReference server : servers.values()) {
shutdownServers(server);
}
servers.clear();
}
+ @Deprecated
Review comment:
If those methods are not used anymore, you can safely remove them. We
are going to release an LTS release with 3.11.0, so we are under heavy
developments, so you can remove methods like these.
##########
File path:
components/camel-solr/src/main/java/org/apache/camel/component/solr/SolrConfiguration.java
##########
@@ -0,0 +1,532 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.solr;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.StringHelper;
+import org.apache.http.client.HttpClient;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.impl.ConcurrentUpdateSolrClient;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
+
+@UriParams
+public class SolrConfiguration implements Cloneable {
+
+ private final SolrScheme solrScheme;
+
+ private boolean useConcurrentUpdateSolrClient;
+ private SolrEndpoint solrEndpoint;
+
+ @UriPath(description = "Hostname and port for the Solr server(s). " +
+ "Multiple hosts can be specified, separated with a
comma. " +
+ "See the solrClient parameter for more information
on the SolrClient used to connect to Solr.")
+ @Metadata(required = true)
+ private String url;
+ @UriParam(defaultValue = "" + SolrConstants.DEFUALT_STREAMING_QUEUE_SIZE)
+ private int streamingQueueSize =
SolrConstants.DEFUALT_STREAMING_QUEUE_SIZE;
+ @UriParam(defaultValue = "" + SolrConstants.DEFAULT_STREAMING_THREAD_COUNT)
+ private int streamingThreadCount =
SolrConstants.DEFAULT_STREAMING_THREAD_COUNT;
+ @UriParam
+ private Integer soTimeout;
+ @UriParam
+ private Integer connectionTimeout;
+ @UriParam(label = "HttpSolrClient")
+ private Boolean followRedirects;
+ @UriParam(label = "HttpSolrClient")
+ private Boolean allowCompression;
+ @UriParam(label = "CloudSolrClient")
+ private String zkHost;
+ @UriParam(label = "CloudSolrClient")
+ private String zkChroot;
+ @UriParam(label = "CloudSolrClient")
+ private String collection;
+ @UriParam
+ private SolrClient solrClient;
+ @UriParam
+ private HttpClient httpClient;
+ @UriParam
+ private String requestHandler;
+ @UriParam(label = "security", secret = true)
+ private String username;
+ @UriParam(label = "security", secret = true)
+ private String password;
+ @UriParam(defaultValue = "false")
+ private boolean autoCommit;
+ @Deprecated
+ @UriParam
+ private Integer maxRetries;
+ @Deprecated
+ @UriParam
+ private Integer defaultMaxConnectionsPerHost;
+ @Deprecated
+ @UriParam
+ private Integer maxTotalConnections;
+
+ public SolrConfiguration(String endpointUri, String remaining) throws
Exception {
+ solrScheme = SolrScheme.SOLR.getFrom(endpointUri);
+ Optional<String> chroot = getChrootFromPath(remaining);
+ if (chroot.isPresent()) {
+ zkChroot = chroot.get();
+ }
+ url = parseHostsFromUrl(remaining, chroot);
+ // validate url
+ getUrlListFrom(url);
+ }
+
+ public SolrScheme getSolrScheme() {
+ return solrScheme;
+ }
+
+ public int getStreamingQueueSize() {
+ return streamingQueueSize;
+ }
+
+ /**
+ * Sets the queue size for the ConcurrentUpdateSolrClient
+ */
+ public void setStreamingQueueSize(int streamingQueueSize) {
+ this.streamingQueueSize = streamingQueueSize;
+ }
+
+ public int getStreamingThreadCount() {
+ return streamingThreadCount;
+ }
+
+ /**
+ * Sets the number of threads for the ConcurrentUpdateSolrClient
+ */
+ public void setStreamingThreadCount(int streamingThreadCount) {
+ this.streamingThreadCount = streamingThreadCount;
+ }
+
+ public Integer getMaxRetries() {
+ return maxRetries;
+ }
+
+ /**
+ * Maximum number of retries to attempt in the event of transient errors
+ */
+ public void setMaxRetries(Integer maxRetries) {
+ this.maxRetries = maxRetries;
+ }
+
+ public Integer getSoTimeout() {
+ return soTimeout;
+ }
+
+ /**
+ * Sets the socket timeout on the SolrClient
+ */
+ public void setSoTimeout(Integer soTimeout) {
+ this.soTimeout = soTimeout;
+ }
+
+ public Integer getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ /**
+ * Sets the connection timeout on the SolrClient
+ */
+ public void setConnectionTimeout(Integer connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ public Integer getDefaultMaxConnectionsPerHost() {
+ return defaultMaxConnectionsPerHost;
+ }
+
+ /**
+ * maxConnectionsPerHost on the underlying HttpConnectionManager
+ */
+ public void setDefaultMaxConnectionsPerHost(Integer
defaultMaxConnectionsPerHost) {
+ this.defaultMaxConnectionsPerHost = defaultMaxConnectionsPerHost;
+ }
+
+ public Integer getMaxTotalConnections() {
+ return maxTotalConnections;
+ }
+
+ /**
+ * maxTotalConnection on the underlying HttpConnectionManager
+ */
+ public void setMaxTotalConnections(Integer maxTotalConnections) {
+ this.maxTotalConnections = maxTotalConnections;
+ }
+
+ public Boolean getFollowRedirects() {
+ return followRedirects;
+ }
+
+ /**
+ * Indicates whether redirects are used to get to the Solr server
+ */
+ public void setFollowRedirects(Boolean followRedirects) {
+ this.followRedirects = followRedirects;
+ }
+
+ public Boolean getAllowCompression() {
+ return allowCompression;
+ }
+
+ /**
+ * Server side must support gzip or deflate for this to have any effect
+ */
+ public void setAllowCompression(Boolean allowCompression) {
+ this.allowCompression = allowCompression;
+ }
+
+ public String getZkHost() {
+ return zkHost;
+ }
+
+ /**
+ * Set the ZooKeeper host(s) urls which the CloudSolrClient uses, e.g.
"zkHost=localhost:8123,localhost:8124".
+ * Optionally add the chroot, e.g.
"zkHost=localhost:8123,localhost:8124/rootformysolr". In case the first part of
+ * the chroot path in the zkHost parameter is set to 'solr' (e.g.
'localhost:8123/solr' or
+ * 'localhost:8123/solr/..'), then that path is not considered as
zookeeper chroot for backward compatibility
+ * reasons (this behaviour can be overridden via zkChroot parameter).
+ */
+ public void setZkHost(String zkHost) {
+ Optional<String> chroot = getChrootFromPath(zkHost);
+ if (chroot.isPresent()) {
+ this.zkChroot = chroot.get();
+ }
+ this.zkHost = parseHostsFromUrl(zkHost, chroot);
+ }
+
+ public String getZkChroot() {
+ return zkChroot;
+ }
+
+ /**
+ * Set the chroot of the zookeeper connection (include the leading slash;
e.g. '/mychroot')
+ */
+ public void setZkChroot(String zkChroot) {
+ this.zkChroot = zkChroot;
+ }
+
+ public String getCollection() {
+ return collection;
+ }
+
+ /**
+ * Set the default collection for SolrCloud
+ */
+ public void setCollection(String collection) {
+ this.collection = collection;
+ }
+
+ public String getRequestHandler() {
+ return requestHandler;
+ }
+
+ /**
+ * Set the request handler to be used
+ */
+ public void setRequestHandler(String requestHandler) {
+ this.requestHandler = requestHandler;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ /**
+ * Sets username for basic auth plugin enabled servers
+ */
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Sets password for basic auth plugin enabled servers
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public boolean isAutoCommit() {
+ return autoCommit;
+ }
+
+ /**
+ * If true, each producer operation will be automatically followed by a
commit
+ */
+ public void setAutoCommit(boolean autoCommit) {
+ this.autoCommit = autoCommit;
+ }
+
+ public SolrClient getSolrClient() {
+ return solrClient;
+ }
+
+ /**
+ * Uses the provided solr client to connect to solr. When this parameter
is not specified, camel applies the
+ * following rules to determine the SolrClient. A CloudSolrClient should
point to a zookeeper endpoint. Other
+ * clients point to a Solr endpoint. 1) when zkHost or zkChroot
(=zookeeper root) parameter is set, then the
+ * CloudSolrClient is used. 2) when multiple hosts are specified in the
uri (separated with a comma), then the
+ * CloudSolrClient (uri scheme is 'solrCloud') or the LBHttpSolrClient
(uri scheme is not 'solrCloud') is used. 3)
+ * when the solr operation is INSERT_STREAMING, then the
ConcurrentUpdateSolrClient is used. 4) otherwise, the
+ * HttpSolrClient is used.
+ */
+ public void setSolrClient(SolrClient solrClient) {
+ this.solrClient = solrClient;
+ }
+
+ public HttpClient getHttpClient() {
+ return httpClient;
+ }
+
+ /**
+ * Sets the http client to be used by the solrClient
+ */
+ public void setHttpClient(HttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ public Boolean getUseConcurrentUpdateSolrClient() {
+ return useConcurrentUpdateSolrClient;
+ }
+
+ void setUseConcurrentUpdateSolrClient(boolean
useConcurrentUpdateSolrClient) {
+ this.useConcurrentUpdateSolrClient = useConcurrentUpdateSolrClient;
+ }
+
+ public SolrEndpoint getSolrEndpoint() {
+ return solrEndpoint;
+ }
+
+ void setSolrEndpoint(SolrEndpoint solrEndpoint) {
+ this.solrEndpoint = solrEndpoint;
+ }
+
+ private String getFirstUrlFrom(String url) throws Exception {
+ return getUrlListFrom(url).get(0);
+ }
+
+ private List<String> getUrlListFrom(String url) throws Exception {
+ // add scheme when required
+ List<String> urlList = Arrays
+ .asList(url.split(","))
+ .stream()
+ .map(s -> solrScheme.getScheme().concat(s))
+ .collect(Collectors.toList());
+ // validate url syntax via parsing in URL instance
+ for (String s : urlList) {
+ try {
+ // solrCloud requires addition of HTTP scheme to be able to
consider it as a valid URL scheme
+ new URL(
+ SolrScheme.SOLRCLOUD.equals(solrScheme) ?
SolrScheme.SOLR.getScheme().concat(s) : s);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Url '%s' not valid for endpoint with uri=%s",
+ s,
+ solrScheme.getUri()));
+ }
+ }
+ return urlList;
+ }
+
+ /**
+ * signature defines parameters deciding whether or not to share the
solrClient - sharing allowed: same signature -
+ * sharing not allowed: different signature
+ */
+ public String getSignature() {
Review comment:
These seems utility methods. I think they should be in a separated class
instead that the configuration class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
