SOLR-7082: Syntactic sugar for metric gathering

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1670176 13f79535-47bb-0310-9956-ffa450edef68

solr/solrj/src/java/org/apache/solr/client/solrj/io/ParallelStream.java

@@ -83,7 +83,9 @@ public class ParallelStream extends CloudSolrStream {
     if(tuple.EOF) {
       Map m = new HashMap();
       m.put("EOF", true);
-      return new Tuple(m);
+      Tuple t = new Tuple(m);
+      t.setMetrics(this.eofTuples);
+      return t;
     }
 
     return tuple;

solr/solrj/src/java/org/apache/solr/client/solrj/io/Tuple.java

@@ -81,10 +81,6 @@ public class Tuple implements Cloneable {
     return (List<Double>)this.fields.get(key);
   }
 
-  public Iterator<Map.Entry> getFields() {
-    return fields.entrySet().iterator();
-  }
-
   public Map getMap() {
     return this.fields;
   }
@@ -95,9 +91,15 @@ public class Tuple implements Cloneable {
 
   public void setMaps(List<Map> maps) {
     this.fields.put("_MAPS_", maps);
-
   }
 
+  public Map<String,Tuple> getMetrics() {
+    return (Map<String,Tuple>)this.fields.get("_METRICS_");
+  }
+
+  public void setMetrics(Map<String, Tuple> metrics) {
+    this.fields.put("_METRICS_", metrics);
+  }
 
   public Tuple clone() {
     HashMap m = new HashMap();

solr/solrj/src/test/org/apache/solr/client/solrj/io/StreamingTest.java

@@ -163,7 +163,7 @@ public class StreamingTest extends AbstractFullDistribZkTestBase {
 
     String zkHost = zkServer.getZkAddress();
 
-    Map paramsA = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_s asc,a_f asc", "partitionKeys", "none");
+    Map paramsA = mapParams("q", "*:*", "fl", "id,a_s,a_i,a_f", "sort", "a_s asc,a_f asc", "partitionKeys", "none");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
     ParallelStream pstream = new ParallelStream(zkHost, "collection1", stream, 2, new AscFieldComp("a_s"));
 
@@ -723,8 +723,17 @@ public class StreamingTest extends AbstractFullDistribZkTestBase {
   protected List<Tuple> getTuples(TupleStream tupleStream) throws IOException {
     tupleStream.open();
     List<Tuple> tuples = new ArrayList();
-    for(Tuple t = tupleStream.read(); !t.EOF; t = tupleStream.read()) {
-      tuples.add(t);
+    for(;;) {
+      Tuple t = tupleStream.read();
+      if(t.EOF) {
+        if(tupleStream instanceof ParallelStream) {
+          ParallelStream p = (ParallelStream) tupleStream;
+          assert(t.getMetrics() == p.getEofTuples()); // Make sure the EOF tuples are properly set on the final EOF tuple
+        }
+        break;
+      } else {
+        tuples.add(t);
+      }
     }
     tupleStream.close();
     return tuples;