You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by th...@apache.org on 2020/08/27 21:28:51 UTC
[lucene-solr] branch reference_impl_dev updated: Re-enable stream
related tests
This is an automated email from the ASF dual-hosted git repository.
thelabdude pushed a commit to branch reference_impl_dev
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
The following commit(s) were added to refs/heads/reference_impl_dev by this push:
new 235c48f Re-enable stream related tests
235c48f is described below
commit 235c48f5c34ec2f0b9c7a63631aa74ef39f924d3
Author: Timothy Potter <th...@gmail.com>
AuthorDate: Thu Aug 27 15:28:31 2020 -0600
Re-enable stream related tests
---
.../OverseerCollectionMessageHandler.java | 3 +-
.../client/solrj/io/graph/GatherNodesStream.java | 2 -
.../client/solrj/io/graph/ShortestPathStream.java | 113 ++++++++++-----------
.../client/solrj/io/graph/GraphExpressionTest.java | 2 +-
.../solr/client/solrj/io/graph/GraphTest.java | 1 -
.../apache/solr/client/solrj/io/sql/JdbcTest.java | 1 -
.../solrj/io/stream/StreamDecoratorTest.java | 34 ++++---
7 files changed, 77 insertions(+), 79 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
index 4e5626c..8c76e5b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/OverseerCollectionMessageHandler.java
@@ -920,9 +920,10 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
latch.countDown();
}
+ // TJP TODO: Getting weird timeout issues when trying to delete a collection that was
+ // created using processAndWait b/c of this latch.await ... need to dig in further.
latch.await(15, TimeUnit.SECONDS); // nocommit - still need a central timeout strat
-
shardHandler.submit(sreq, replica, sreq.params);
ShardResponse srsp;
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/GatherNodesStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/GatherNodesStream.java
index 3e15ee2..7e42182 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/GatherNodesStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/GatherNodesStream.java
@@ -607,8 +607,6 @@ public class GatherNodesStream extends TupleStream implements Expressible {
out = traversal.iterator();
} catch(Exception e) {
throw new RuntimeException(e);
- } finally {
- ExecutorUtil.shutdownAndAwaitTermination(threadPool);
}
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/ShortestPathStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/ShortestPathStream.java
index 5b1fea2..28f00c1 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/ShortestPathStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/ShortestPathStream.java
@@ -299,48 +299,56 @@ public class ShortestPathStream extends TupleStream implements Expressible {
List<Edge> targets = new ArrayList();
ExecutorService threadPool = null;
- try {
-
- threadPool = ParWork.getExecutor();
-
- //Breadth first search
- TRAVERSE:
- while (targets.size() == 0 && depth < maxDepth) {
- Set<String> nodes = visited.keySet();
- Iterator<String> it = nodes.iterator();
- nextVisited = new HashMap();
- int batchCount = 0;
- List<String> queryNodes = new ArrayList();
- List<Future> futures = new ArrayList();
- JOIN:
- //Queue up all the batches
- while (it.hasNext()) {
- String node = it.next();
- queryNodes.add(node);
- ++batchCount;
- if (batchCount == joinBatchSize || !it.hasNext()) {
- try {
- JoinRunner joinRunner = new JoinRunner(queryNodes);
- Future<List<Edge>> future = threadPool.submit(joinRunner);
- futures.add(future);
- } catch (Exception e) {
- ParWork.propegateInterrupt(e);
- throw new RuntimeException(e);
- }
- batchCount = 0;
- queryNodes = new ArrayList();
+ threadPool = ParWork.getExecutor();
+
+ //Breadth first search
+ TRAVERSE:
+ while (targets.size() == 0 && depth < maxDepth) {
+ Set<String> nodes = visited.keySet();
+ Iterator<String> it = nodes.iterator();
+ nextVisited = new HashMap();
+ int batchCount = 0;
+ List<String> queryNodes = new ArrayList();
+ List<Future> futures = new ArrayList();
+ JOIN:
+ //Queue up all the batches
+ while (it.hasNext()) {
+ String node = it.next();
+ queryNodes.add(node);
+ ++batchCount;
+ if (batchCount == joinBatchSize || !it.hasNext()) {
+ try {
+ JoinRunner joinRunner = new JoinRunner(queryNodes);
+ Future<List<Edge>> future = threadPool.submit(joinRunner);
+ futures.add(future);
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new RuntimeException(e);
}
+ batchCount = 0;
+ queryNodes = new ArrayList();
}
+ }
- try {
- //Process the batches as they become available
- OUTER:
- for (Future<List<Edge>> future : futures) {
- List<Edge> edges = future.get();
- INNER:
- for (Edge edge : edges) {
- if (toNode.equals(edge.to)) {
- targets.add(edge);
+ try {
+ //Process the batches as they become available
+ OUTER:
+ for (Future<List<Edge>> future : futures) {
+ List<Edge> edges = future.get();
+ INNER:
+ for (Edge edge : edges) {
+ if (toNode.equals(edge.to)) {
+ targets.add(edge);
+ if(nextVisited.containsKey(edge.to)) {
+ List<String> parents = nextVisited.get(edge.to);
+ parents.add(edge.from);
+ } else {
+ List<String> parents = new ArrayList();
+ parents.add(edge.from);
+ nextVisited.put(edge.to, parents);
+ }
+ } else {
+ if (!cycle(edge.to, allVisited)) {
if(nextVisited.containsKey(edge.to)) {
List<String> parents = nextVisited.get(edge.to);
parents.add(edge.from);
@@ -349,31 +357,18 @@ public class ShortestPathStream extends TupleStream implements Expressible {
parents.add(edge.from);
nextVisited.put(edge.to, parents);
}
- } else {
- if (!cycle(edge.to, allVisited)) {
- if(nextVisited.containsKey(edge.to)) {
- List<String> parents = nextVisited.get(edge.to);
- parents.add(edge.from);
- } else {
- List<String> parents = new ArrayList();
- parents.add(edge.from);
- nextVisited.put(edge.to, parents);
- }
- }
}
}
}
- } catch (Exception e) {
- ParWork.propegateInterrupt(e);
- throw new RuntimeException(e);
}
-
- allVisited.add(nextVisited);
- visited = nextVisited;
- ++depth;
+ } catch (Exception e) {
+ ParWork.propegateInterrupt(e);
+ throw new RuntimeException(e);
}
- } finally {
- ExecutorUtil.shutdownAndAwaitTermination(threadPool);
+
+ allVisited.add(nextVisited);
+ visited = nextVisited;
+ ++depth;
}
Set<String> finalPaths = new HashSet();
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphExpressionTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphExpressionTest.java
index bf0c44d..7d303f4 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphExpressionTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphExpressionTest.java
@@ -73,7 +73,7 @@ import org.junit.Test;
**/
@Slow
-@Ignore // nocommit flakey
+//@Ignore // nocommit flakey
@LuceneTestCase.SuppressCodecs({"Lucene3x", "Lucene40","Lucene41","Lucene42","Lucene45"})
public class GraphExpressionTest extends SolrCloudTestCase {
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphTest.java
index 005b792..7d59c08 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/graph/GraphTest.java
@@ -45,7 +45,6 @@ import org.junit.Test;
@LuceneTestCase.Slow
@LuceneTestCase.SuppressCodecs({"Lucene3x", "Lucene40","Lucene41","Lucene42","Lucene45"})
-@Ignore // nocommit - debug
public class GraphTest extends SolrCloudTestCase {
private static final String COLLECTION = "collection1";
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/sql/JdbcTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/sql/JdbcTest.java
index 557f491..8ea1562 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/sql/JdbcTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/sql/JdbcTest.java
@@ -54,7 +54,6 @@ import org.junit.Test;
@Slow
@LuceneTestCase.SuppressCodecs({"Lucene3x", "Lucene40", "Lucene41", "Lucene42", "Lucene45"})
-@Ignore // nocommit - debug
public class JdbcTest extends SolrCloudTestCase {
private static final String COLLECTIONORALIAS = "collection1";
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
index 51a1ace..f389251 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java
@@ -58,7 +58,6 @@ import org.apache.solr.client.solrj.io.stream.metrics.SumMetric;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
-import org.apache.solr.cloud.AbstractDistribZkTestBase;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.cloud.ClusterState;
@@ -74,7 +73,6 @@ import org.junit.Test;
@Slow
@SolrTestCaseJ4.SuppressSSL
@LuceneTestCase.SuppressCodecs({"Lucene3x", "Lucene40","Lucene41","Lucene42","Lucene45"})
-@Ignore // nocommit debug
public class StreamDecoratorTest extends SolrCloudTestCase {
private static final String COLLECTIONORALIAS = "collection1";
@@ -962,6 +960,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
@Test
+ @Nightly // too slow for non-Nightly
public void testDaemonStream() throws Exception {
new UpdateRequest()
@@ -2766,6 +2765,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
@Test
+ @Ignore // nocommit ~ TJP
public void testParallelUpdateStream() throws Exception {
CollectionAdminRequest.createCollection("parallelDestinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
@@ -2864,6 +2864,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
@Test
+ @Ignore // nocommit ~ TJP
public void testParallelDaemonUpdateStream() throws Exception {
CollectionAdminRequest.createCollection("parallelDestinationCollection1", "conf", 2, 1).process(cluster.getSolrClient());
@@ -3037,6 +3038,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
@Test
+ @Ignore // nocommit ~ TJP
public void testParallelTerminatingDaemonUpdateStream() throws Exception {
Assume.assumeTrue(!useAlias);
@@ -3221,9 +3223,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
-
-
@Test
+ @Ignore // nocommit ~ TJP
public void testCommitStream() throws Exception {
CollectionAdminRequest.createCollection("destinationCollection", "conf", 2, 1).process(cluster.getSolrClient());
@@ -3860,19 +3861,22 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
}
@Test
+ @Ignore // nocommit ~ TJP ~ slow and/or flakey, esp. around collection creation / deletion
public void testExecutorStream() throws Exception {
- CollectionAdminRequest.createCollection("workQueue", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("workQueue", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("mainCorpus", "conf", 2, 1).process(cluster.getSolrClient());
+ CollectionAdminRequest.createCollection("destination", "conf", 2, 1).process(cluster.getSolrClient());
+
cluster.waitForActiveCollection("workQueue", 2, 2);
- CollectionAdminRequest.createCollection("mainCorpus", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("mainCorpus", 2, 2);
- CollectionAdminRequest.createCollection("destination", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
cluster.waitForActiveCollection("destination", 2, 2);
UpdateRequest workRequest = new UpdateRequest();
UpdateRequest dataRequest = new UpdateRequest();
+ final int numDocs = 500;
- for (int i = 0; i < 500; i++) {
+ for (int i = 0; i < numDocs; i++) {
workRequest.add(id, String.valueOf(i), "expr_s", "update(destination, batchSize=50, search(mainCorpus, q=id:"+i+", rows=1, sort=\"id asc\", fl=\"id, body_t, field_i\"))");
dataRequest.add(id, String.valueOf(i), "body_t", "hello world "+i, "field_i", Integer.toString(i));
}
@@ -3880,7 +3884,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
workRequest.commit(cluster.getSolrClient(), "workQueue");
dataRequest.commit(cluster.getSolrClient(), "mainCorpus");
- String url = cluster.getJettySolrRunners().get(0).getBaseUrl().toString() + "/destination";
+ String url = cluster.getJettySolrRunners().get(0).getBaseUrl() + "/destination";
TupleStream executorStream;
ModifiableSolrParams paramsLoc;
@@ -3908,8 +3912,8 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
SolrStream solrStream = new SolrStream(url, paramsLoc);
List<Tuple> tuples = getTuples(solrStream);
- assertTrue(tuples.size() == 500);
- for(int i=0; i<500; i++) {
+ assertTrue(tuples.size() == numDocs);
+ for(int i=0; i<numDocs; i++) {
Tuple tuple = tuples.get(i);
long ivalue = tuple.getLong("field_i");
String body = tuple.getString("body_t");
@@ -3919,6 +3923,7 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
solrStream.close();
clientCache.close();
+
CollectionAdminRequest.deleteCollection("workQueue").process(cluster.getSolrClient());
CollectionAdminRequest.deleteCollection("mainCorpus").process(cluster.getSolrClient());
CollectionAdminRequest.deleteCollection("destination").process(cluster.getSolrClient());
@@ -3926,12 +3931,13 @@ public class StreamDecoratorTest extends SolrCloudTestCase {
@Test
+ @Ignore // nocommit ~ hangs during shutdown
public void testParallelExecutorStream() throws Exception {
- CollectionAdminRequest.createCollection("workQueue1", "conf", 2, 1).processAndWait(cluster.getSolrClient(),DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("workQueue1", "conf", 2, 1).process(cluster.getSolrClient());
- CollectionAdminRequest.createCollection("mainCorpus1", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("mainCorpus1", "conf", 2, 1).process(cluster.getSolrClient());
- CollectionAdminRequest.createCollection("destination1", "conf", 2, 1).processAndWait(cluster.getSolrClient(), DEFAULT_TIMEOUT);
+ CollectionAdminRequest.createCollection("destination1", "conf", 2, 1).process(cluster.getSolrClient());
cluster.waitForActiveCollection("workQueue1", 2, 2);
cluster.waitForActiveCollection("mainCorpus1", 2, 2);