You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ma...@apache.org on 2020/08/27 22:37:45 UTC
[lucene-solr] 05/10: Re-enable StreamExpressionTest and fix usage
of ParWork Executor, it should no longer be closed by stream classes.
This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch reference_impl
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 1652056a25b484a0fa2ead0a0cae6c73ec42165d
Author: Timothy Potter <th...@gmail.com>
AuthorDate: Wed Aug 26 15:08:21 2020 -0600
Re-enable StreamExpressionTest and fix usage of ParWork Executor, it should no longer be closed by stream classes.
---
.../client/solrj/io/stream/CloudSolrStream.java | 2 +-
.../client/solrj/io/stream/DeepRandomStream.java | 31 ++++++-------------
.../client/solrj/io/stream/ExecutorStream.java | 6 +---
.../solrj/io/stream/FeaturesSelectionStream.java | 11 +++----
.../client/solrj/io/stream/ParallelListStream.java | 34 +++++++++-----------
.../solrj/io/stream/SignificantTermsStream.java | 6 ++--
.../solr/client/solrj/io/stream/SolrStream.java | 14 +++------
.../client/solrj/io/stream/TextLogitStream.java | 12 +++-----
.../solr/client/solrj/io/stream/TopicStream.java | 36 +---------------------
.../solrj/io/stream/StreamExpressionTest.java | 2 +-
10 files changed, 43 insertions(+), 111 deletions(-)
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
index b49a820..5507cf2 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
@@ -394,7 +394,7 @@ public class CloudSolrStream extends TupleStream implements Expressible {
}
}
- private void openStreams() throws IOException {
+ protected void openStreams() throws IOException {
final ExecutorService service = ParWork.getExecutor();
List<Future<TupleWrapper>> futures =
solrStreams.stream().map(ss -> service.submit(new StreamOpener((SolrStream)ss, comp))).collect(Collectors.toList());
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DeepRandomStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DeepRandomStream.java
index e215290..1e0f8f0 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DeepRandomStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DeepRandomStream.java
@@ -51,8 +51,6 @@ import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
-import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import static org.apache.solr.common.params.CommonParams.DISTRIB;
import static org.apache.solr.common.params.CommonParams.ROWS;
@@ -348,28 +346,19 @@ public class DeepRandomStream extends TupleStream implements Expressible {
}
private void openStreams() throws IOException {
- ExecutorService service = ParWork.getExecutor();
+ final ExecutorService service = ParWork.getExecutor();
+ List<Future<TupleWrapper>> futures =
+ solrStreams.stream().map(ss -> service.submit(new StreamOpener((SolrStream)ss, comp))).collect(Collectors.toList());
try {
- List<Future<TupleWrapper>> futures = new ArrayList();
- for (TupleStream solrStream : solrStreams) {
- StreamOpener so = new StreamOpener((SolrStream) solrStream, comp);
- Future<TupleWrapper> future = service.submit(so);
- futures.add(future);
- }
-
- try {
- for (Future<TupleWrapper> f : futures) {
- TupleWrapper w = f.get();
- if (w != null) {
- tuples.add(w);
- }
+ for (Future<TupleWrapper> f : futures) {
+ TupleWrapper w = f.get();
+ if (w != null) {
+ tuples.add(w);
}
- } catch (Exception e) {
- ParWork.propegateInterrupt(e);
- throw new IOException(e);
}
- } finally {
- service.shutdown();
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new IOException(e);
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
index 1e0824b..cf6d8ad 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExecutorStream.java
@@ -24,7 +24,6 @@ import java.util.List;
import java.util.Locale;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
@@ -37,8 +36,6 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParamete
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.ParWork;
-import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -141,11 +138,10 @@ public class ExecutorStream extends TupleStream implements Expressible {
public void close() throws IOException {
stream.close();
- ExecutorUtil.shutdownAndAwaitTermination(executorService);
}
public Tuple read() throws IOException {
- ArrayBlockingQueue<Tuple> queue = new ArrayBlockingQueue(10000);
+ ArrayBlockingQueue<Tuple> queue = new ArrayBlockingQueue<>(10000);
while(true) {
Tuple tuple = stream.read();
if (!tuple.EOF) {
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/FeaturesSelectionStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/FeaturesSelectionStream.java
index d1a9c0a..8c321cf 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/FeaturesSelectionStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/FeaturesSelectionStream.java
@@ -35,9 +35,7 @@ import java.util.concurrent.Future;
import java.util.stream.Stream;
import org.apache.solr.client.solrj.impl.CloudHttp2SolrClient;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
@@ -58,9 +56,7 @@ import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import static org.apache.solr.common.params.CommonParams.DISTRIB;
import static org.apache.solr.common.params.CommonParams.ID;
@@ -313,9 +309,10 @@ public class FeaturesSelectionStream extends TupleStream implements Expressible{
cache.close();
}
- if (executorService != null) {
- executorService.shutdown();
- }
+ // We don't own the executorService, so no shutdown
+ //if (executorService != null) {
+ // executorService.shutdown();
+ //}
}
/** Return the stream sort - ie, the order in which records are returned */
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelListStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelListStream.java
index 10b8342..6562b33 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelListStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelListStream.java
@@ -33,8 +33,6 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.ParWork;
-import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
public class ParallelListStream extends TupleStream implements Expressible {
@@ -136,26 +134,22 @@ public class ParallelListStream extends TupleStream implements Expressible {
private void openStreams() throws IOException {
ExecutorService service = ParWork.getExecutor();
- try {
- List<Future<StreamIndex>> futures = new ArrayList();
- int i=0;
- for (TupleStream tupleStream : streams) {
- StreamOpener so = new StreamOpener(new StreamIndex(tupleStream, i++));
- Future<StreamIndex> future = service.submit(so);
- futures.add(future);
- }
+ List<Future<StreamIndex>> futures = new ArrayList();
+ int i=0;
+ for (TupleStream tupleStream : streams) {
+ StreamOpener so = new StreamOpener(new StreamIndex(tupleStream, i++));
+ Future<StreamIndex> future = service.submit(so);
+ futures.add(future);
+ }
- try {
- for (Future<StreamIndex> f : futures) {
- StreamIndex streamIndex = f.get();
- this.streams[streamIndex.getIndex()] = streamIndex.getTupleStream();
- }
- } catch (Exception e) {
- ParWork.propegateInterrupt(e);
- throw new IOException(e);
+ try {
+ for (Future<StreamIndex> f : futures) {
+ StreamIndex streamIndex = f.get();
+ this.streams[streamIndex.getIndex()] = streamIndex.getTupleStream();
}
- } finally {
- ExecutorUtil.shutdownAndAwaitTermination(service);
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new IOException(e);
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SignificantTermsStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SignificantTermsStream.java
index c08c3bb..28169de 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SignificantTermsStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SignificantTermsStream.java
@@ -32,7 +32,6 @@ import java.util.concurrent.Future;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
@@ -49,9 +48,7 @@ import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.ParWork;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import static org.apache.solr.common.params.CommonParams.DISTRIB;
@@ -266,7 +263,8 @@ public class SignificantTermsStream extends TupleStream implements Expressible{
cache.close();
}
- executorService.shutdown();
+ // I don't own the executorService anymore, so no shutting it down
+ // executorService.shutdown();
}
/** Return the stream sort - ie, the order in which records are returned */
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
index 6d43f5f..e879325 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
@@ -16,11 +16,8 @@
*/
package org.apache.solr.client.solrj.io.stream;
-import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -30,9 +27,7 @@ import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.impl.BaseHttpSolrClient;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.InputStreamResponseParser;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.io.Tuple;
@@ -68,7 +63,6 @@ public class SolrStream extends TupleStream {
private Map<String, String> fieldMappings;
private transient TupleStreamParser tupleStreamParser;
private transient Http2SolrClient client;
- private transient Closeable closeableHttp2SolrClient;
private transient SolrClientCache cache;
private String slice;
private long checkpoint = -1;
@@ -118,10 +112,8 @@ public class SolrStream extends TupleStream {
public void open() throws IOException {
if(cache == null) {
client = new Http2SolrClient.Builder(baseUrl).markInternalRequest().build();
- closeableHttp2SolrClient = client;
} else {
client = cache.getHttpSolrClient(baseUrl);
- closeableHttp2SolrClient = null;
}
try {
@@ -199,8 +191,10 @@ public class SolrStream extends TupleStream {
}
IOUtils.closeQuietly(tupleStreamParser);
tupleStreamParser.close();
- if (closeableHttp2SolrClient != null) {
- closeableHttp2SolrClient.close();
+
+ // if the cache is null, then we opened the client
+ if (cache == null && client != null) {
+ client.close();
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TextLogitStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TextLogitStream.java
index a0422d4..190b228 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TextLogitStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TextLogitStream.java
@@ -35,9 +35,7 @@ import java.util.concurrent.Future;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.impl.CloudHttp2SolrClient;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.io.ClassificationEvaluation;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.io.Tuple;
@@ -59,9 +57,7 @@ import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
-import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import static org.apache.solr.common.params.CommonParams.DISTRIB;
import static org.apache.solr.common.params.CommonParams.ID;
@@ -399,9 +395,11 @@ public class TextLogitStream extends TupleStream implements Expressible {
cache.close();
}
- if (executorService != null) {
- executorService.shutdown();
- }
+ // We don't own the executorService, so don't shutdown
+ //if (executorService != null) {
+ // executorService.shutdown();
+ //}
+
termsStream.close();
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TopicStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TopicStream.java
index fd69866..a72db06 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TopicStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TopicStream.java
@@ -25,19 +25,15 @@ import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
-import java.util.Optional;
import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.impl.CloudHttp2SolrClient;
-import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
import org.apache.solr.client.solrj.io.comp.FieldComparator;
@@ -60,8 +56,6 @@ import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
-import org.apache.solr.common.util.ExecutorUtil;
-import org.apache.solr.common.util.SolrNamedThreadFactory;
import static org.apache.solr.common.params.CommonParams.DISTRIB;
import static org.apache.solr.common.params.CommonParams.ID;
@@ -311,34 +305,6 @@ public class TopicStream extends CloudSolrStream implements Expressible {
openStreams();
}
-
- private void openStreams() throws IOException {
-
- ExecutorService service = ParWork.getExecutor();
- try {
- List<Future<TupleWrapper>> futures = new ArrayList();
- for (TupleStream solrStream : solrStreams) {
- StreamOpener so = new StreamOpener((SolrStream) solrStream, comp);
- Future<TupleWrapper> future = service.submit(so);
- futures.add(future);
- }
-
- try {
- for (Future<TupleWrapper> f : futures) {
- TupleWrapper w = f.get();
- if (w != null) {
- tuples.add(w);
- }
- }
- } catch (Exception e) {
- ParWork.propegateInterrupt(e);
- throw new IOException(e);
- }
- } finally {
- ExecutorUtil.shutdownAndAwaitTermination(service);
- }
- }
-
public void close() throws IOException {
try {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java
index 9f413de..6efafa8 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java
@@ -67,7 +67,6 @@ import org.junit.Test;
@Slow
@SolrTestCaseJ4.SuppressSSL
@LuceneTestCase.SuppressCodecs({"Lucene3x", "Lucene40","Lucene41","Lucene42","Lucene45"})
-@Ignore // nocommit debug
public class StreamExpressionTest extends SolrCloudTestCase {
private static final String COLLECTIONORALIAS = "collection1";
@@ -3135,6 +3134,7 @@ public class StreamExpressionTest extends SolrCloudTestCase {
@Test
+ @Nightly // this test creates a lot of data and is slow
public void testSignificantTermsStream() throws Exception {
UpdateRequest updateRequest = new UpdateRequest();