diff --git a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/interactor/GetQueryResultStream.java b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/interactor/GetQueryResultStream.java
index 951d060..0e70391 100644
--- a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/interactor/GetQueryResultStream.java
+++ b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/interactor/GetQueryResultStream.java
@@ -37,18 +37,20 @@
     /**
      * Stream all of the results that have been produced by a query.
      *
+     * @param ryaInstance - The name of the Rya Instance the query is for. (not null)
      * @param queryId - Indicates which query results to stream. (not null)
      * @return A {@link QueryResultStream} that starts with the first result that was ever produced.
      * @throws RyaStreamsException Could not create the result stream.
      */
-    public QueryResultStream<T> fromStart(UUID queryId) throws RyaStreamsException;
+    public QueryResultStream<T> fromStart(String ryaInstance, UUID queryId) throws RyaStreamsException;
 
     /**
      * Stream results that have been produced by a query after this method was invoked.
      *
+     * @param ryaInstance - The name of the Rya Instance the query is for. (not null)
      * @param queryId - Indicates which query results to stream. (not null)
      * @return A {@link QueryResultStream} that only returns results that were produced after this method is invoked.
      * @throws RyaStreamsException Could not create the result stream.
      */
-    public QueryResultStream<T> fromNow(UUID queryId) throws RyaStreamsException;
+    public QueryResultStream<T> fromNow(String ryaInstance, UUID queryId) throws RyaStreamsException;
 }
\ No newline at end of file
diff --git a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/RunQueryCommand.java b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/RunQueryCommand.java
index 7b311f6..f9a9458 100644
--- a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/RunQueryCommand.java
+++ b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/RunQueryCommand.java
@@ -135,7 +135,7 @@
                 // Make sure the topics required by the application exists for the specified Rya instances.
                 final Set<String> topics = new HashSet<>();
                 topics.add( KafkaTopics.statementsTopic(params.ryaInstance) );
-                topics.add( KafkaTopics.queryResultsTopic(queryId) );
+                topics.add( KafkaTopics.queryResultsTopic(params.ryaInstance, queryId) );
                 KafkaTopics.createTopics(params.zookeeperServers, topics, 1, 1);
 
                 // Run the query that uses those topics.
@@ -143,7 +143,7 @@
                         params.kafkaIP,
                         params.kafkaPort,
                         KafkaTopics.statementsTopic(params.ryaInstance),
-                        KafkaTopics.queryResultsTopic(queryId),
+                        KafkaTopics.queryResultsTopic(params.ryaInstance, queryId),
                         queryRepo,
                         new TopologyFactory());
                 runQuery.run(queryId);
diff --git a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
index 8ae4e08..783aedc 100644
--- a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
+++ b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/StreamResultsCommand.java
@@ -1,4 +1,4 @@
-/**
+/*
  * 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
@@ -20,18 +20,11 @@
 
 import static java.util.Objects.requireNonNull;
 
-import java.io.OutputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.rya.api.model.VisibilityBindingSet;
-import org.apache.rya.api.model.VisibilityStatement;
-import org.apache.rya.api.utils.QueryInvestigator;
 import org.apache.rya.streams.api.entity.QueryResultStream;
 import org.apache.rya.streams.api.entity.StreamsQuery;
 import org.apache.rya.streams.api.interactor.GetQueryResultStream;
@@ -39,13 +32,13 @@
 import org.apache.rya.streams.api.queries.QueryChangeLog;
 import org.apache.rya.streams.api.queries.QueryRepository;
 import org.apache.rya.streams.client.RyaStreamsCommand;
-import org.apache.rya.streams.client.util.QueryResultsOutputUtil;
 import org.apache.rya.streams.kafka.KafkaTopics;
 import org.apache.rya.streams.kafka.interactor.KafkaGetQueryResultStream;
 import org.apache.rya.streams.kafka.queries.KafkaQueryChangeLogFactory;
 import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
 import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer;
 import org.openrdf.query.MalformedQueryException;
+import org.openrdf.query.algebra.Reduced;
 import org.openrdf.query.algebra.TupleExpr;
 import org.openrdf.query.parser.sparql.SPARQLParser;
 
@@ -72,20 +65,14 @@
         @Parameter(names = {"--queryId", "-q"}, required = true, description = "The query whose results will be streamed to the console.")
         private String queryId;
 
-        @Parameter(names = {"--file", "-f"}, required = false, description = "If provided, the output file the results will stream into.")
-        private String outputPath;
-
         @Override
         public String toString() {
             final StringBuilder parameters = new StringBuilder();
             parameters.append(super.toString());
 
             if (!Strings.isNullOrEmpty(queryId)) {
-                parameters.append("\tQuery ID: " + queryId + "\n");
-            }
-
-            if(!Strings.isNullOrEmpty(outputPath)) {
-                parameters.append("\tOutput Path: " + outputPath + "\n");
+                parameters.append("\tQuery ID: " + queryId);
+                parameters.append("\n");
             }
 
             return parameters.toString();
@@ -173,12 +160,10 @@
         });
 
         // Build the interactor based on the type of result the query produces.
-        boolean isStatementResults = false;
-
         final GetQueryResultStream<?> getQueryResultStream;
         try {
-            isStatementResults = QueryInvestigator.isConstruct(sparql) | QueryInvestigator.isInsertWhere(sparql);
-            if(isStatementResults) {
+            final TupleExpr tupleExpr = new SPARQLParser().parseQuery(sparql, null).getTupleExpr();
+            if(tupleExpr instanceof Reduced) {
                 getQueryResultStream = new KafkaGetQueryResultStream<>(params.kafkaIP, params.kafkaPort, VisibilityStatementDeserializer.class);
             } else {
                 getQueryResultStream = new KafkaGetQueryResultStream<>(params.kafkaIP, params.kafkaPort, VisibilityBindingSetDeserializer.class);
@@ -187,22 +172,12 @@
             throw new ExecutionException("Could not parse the SPARQL for the query: " + sparql, e);
         }
 
-        // Iterate through the results and print them to the configured output mechanism.
-        try (final QueryResultStream<?> resultsStream = getQueryResultStream.fromStart(queryId)) {
-            final TupleExpr tupleExpr = new SPARQLParser().parseQuery(sparql, null).getTupleExpr();
-            if(params.outputPath != null) {
-                final Path file = Paths.get(params.outputPath);
-                try (final OutputStream out = Files.newOutputStream(file)) {
-                    if(isStatementResults) {
-                        final QueryResultStream<VisibilityStatement> stmtStream = (QueryResultStream<VisibilityStatement>) resultsStream;
-                        QueryResultsOutputUtil.toNtriplesFile(out, stmtStream, finished);
-                    } else {
-                        final QueryResultStream<VisibilityBindingSet> bsStream = (QueryResultStream<VisibilityBindingSet>) resultsStream;
-                        QueryResultsOutputUtil.toBindingSetJSONFile(out, tupleExpr, bsStream, finished);
-                    }
+        // Iterate through the results and print them to the console until the program or the stream ends.
+        try (final QueryResultStream<?> stream = getQueryResultStream.fromStart(params.ryaInstance, queryId)) {
+            while(!finished.get()) {
+                for(final Object result : stream.poll(1000)) {
+                    System.out.println(result);
                 }
-            } else {
-                streamToSystemOut(resultsStream, finished);
             }
         } catch (final Exception e) {
             System.err.println("Error while reading the results from the stream.");
@@ -210,15 +185,4 @@
             System.exit(1);
         }
     }
-
-    private static void streamToSystemOut(final QueryResultStream<?> stream, final AtomicBoolean shutdownSignal) throws Exception {
-        requireNonNull(stream);
-        requireNonNull(shutdownSignal);
-
-        while(!shutdownSignal.get()) {
-            for(final Object result : stream.poll(1000)) {
-                System.out.println(result);
-            }
-        }
-    }
-}
\ No newline at end of file
+}
diff --git a/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/command/RunQueryCommandIT.java b/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/command/RunQueryCommandIT.java
index 5d63f32..176b920 100644
--- a/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/command/RunQueryCommandIT.java
+++ b/extras/rya.streams/client/src/test/java/org/apache/rya/streams/client/command/RunQueryCommandIT.java
@@ -184,7 +184,7 @@
             loadStatements.fromCollection(statements);
 
             // Read the output of the streams program.
-            final String resultsTopic = KafkaTopics.queryResultsTopic(sQuery.getQueryId());
+            final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, sQuery.getQueryId());
             resultConsumer.subscribe( Lists.newArrayList(resultsTopic) );
             results = KafkaTestUtil.pollForResults(500, 6, 3, resultConsumer);
         } finally {
diff --git a/extras/rya.streams/geo/src/test/java/org/apache/rya/streams/kafka/processors/filter/GeoFilterIT.java b/extras/rya.streams/geo/src/test/java/org/apache/rya/streams/kafka/processors/filter/GeoFilterIT.java
index c090afa..17a290a 100644
--- a/extras/rya.streams/geo/src/test/java/org/apache/rya/streams/kafka/processors/filter/GeoFilterIT.java
+++ b/extras/rya.streams/geo/src/test/java/org/apache/rya/streams/kafka/processors/filter/GeoFilterIT.java
@@ -88,7 +88,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/aggregation/AggregationProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/aggregation/AggregationProcessorIT.java
index a4e4a3e..2a1e760 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/aggregation/AggregationProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/aggregation/AggregationProcessorIT.java
@@ -34,7 +34,6 @@
 import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
 import org.apache.rya.streams.kafka.topology.TopologyFactory;
 import org.apache.rya.test.kafka.KafkaTestInstanceRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.openrdf.model.ValueFactory;
@@ -90,7 +89,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -137,7 +136,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -183,7 +182,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -233,7 +232,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -283,7 +282,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -371,7 +370,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
@@ -380,9 +379,7 @@
         RyaStreamsTestUtil.runStreamProcessingTest(kafka, statementsTopic, resultsTopic, builder, statements, expected, VisibilityBindingSetDeserializer.class);
     }
 
-    // Ignored because this test is kind of flakey.
     @Test
-    @Ignore
     public void multipleAggregations() throws Exception {
         // A query that figures out what the youngest and oldest ages are across all people.
         final String sparql =
@@ -436,7 +433,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final TopologyBuilder builder = new TopologyFactory().build(sparql, statementsTopic, resultsTopic, new RandomUUIDFactory());
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/FilterProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/FilterProcessorIT.java
index aaa67ea..7fb228a 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/FilterProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/FilterProcessorIT.java
@@ -54,7 +54,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/TemporalFilterIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/TemporalFilterIT.java
index 6e27669..11637b7 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/TemporalFilterIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/filter/TemporalFilterIT.java
@@ -85,7 +85,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
@@ -119,7 +119,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
@@ -153,7 +153,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
@@ -187,7 +187,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Get the RDF model objects that will be used to build the query.
         final String sparql =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java
index bdb9be6..5f09372 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/join/JoinProcessorIT.java
@@ -69,7 +69,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query =
@@ -120,7 +120,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query =
@@ -171,7 +171,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query =
@@ -228,7 +228,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query =
@@ -269,7 +269,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/MultiProjectionProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/MultiProjectionProcessorIT.java
index a560294..c6fd1cf 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/MultiProjectionProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/MultiProjectionProcessorIT.java
@@ -28,7 +28,6 @@
 import org.apache.rya.api.model.VisibilityStatement;
 import org.apache.rya.streams.kafka.KafkaTopics;
 import org.apache.rya.streams.kafka.RyaStreamsTestUtil;
-import org.apache.rya.streams.kafka.processors.projection.MultiProjectionProcessorSupplier;
 import org.apache.rya.streams.kafka.processors.projection.MultiProjectionProcessorSupplier.MultiProjectionProcessor;
 import org.apache.rya.streams.kafka.serialization.VisibilityStatementDeserializer;
 import org.apache.rya.streams.kafka.topology.TopologyFactory;
@@ -54,7 +53,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Create a topology for the Query that will be tested.
         final String sparql =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/ProjectionProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/ProjectionProcessorIT.java
index 322bba9..f53f2c4 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/ProjectionProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/projection/ProjectionProcessorIT.java
@@ -30,7 +30,6 @@
 import org.apache.rya.api.model.VisibilityStatement;
 import org.apache.rya.streams.kafka.KafkaTopics;
 import org.apache.rya.streams.kafka.RyaStreamsTestUtil;
-import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier;
 import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier.StatementPatternProcessor;
 import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
 import org.apache.rya.streams.kafka.topology.TopologyFactory;
@@ -57,7 +56,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Create a topology for the Query that will be tested.
         final String sparql =
diff --git a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/sp/StatementPatternProcessorIT.java b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/sp/StatementPatternProcessorIT.java
index ba11e57..fd0a48d 100644
--- a/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/sp/StatementPatternProcessorIT.java
+++ b/extras/rya.streams/integration/src/test/java/org/apache/rya/streams/kafka/processors/sp/StatementPatternProcessorIT.java
@@ -30,7 +30,6 @@
 import org.apache.rya.api.model.VisibilityStatement;
 import org.apache.rya.streams.kafka.KafkaTopics;
 import org.apache.rya.streams.kafka.RyaStreamsTestUtil;
-import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier;
 import org.apache.rya.streams.kafka.processors.StatementPatternProcessorSupplier.StatementPatternProcessor;
 import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
 import org.apache.rya.streams.kafka.topology.TopologyFactory;
@@ -55,7 +54,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query = "SELECT * WHERE { ?person <urn:talksTo> ?otherPerson }";
@@ -85,7 +84,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query = "SELECT * WHERE { ?person <urn:talksTo> ?otherPerson }";
@@ -123,7 +122,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query = "SELECT * WHERE { "
@@ -157,7 +156,7 @@
         final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // Setup a topology.
         final String query = "SELECT * WHERE { "
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
index 989799a..f0cc842 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java
@@ -62,7 +62,7 @@
      * @param changeLogTopic - The topic to evaluate. (not null)
      * @return If the topic is well formatted, then the Rya instance name that was part of the topic name.
      */
-    public static Optional<String> getRyaInstance(final String changeLogTopic) {
+    public static Optional<String> getRyaInstanceFromQueryChangeLog(final String changeLogTopic) {
         requireNonNull(changeLogTopic);
 
         // Return absent if the provided topic does not represent a query change log topic.
@@ -93,9 +93,31 @@
      * @param queryId - The id of the query the topic is for.
      * @return The name of the Kafka topic.
      */
-    public static String queryResultsTopic(final UUID queryId) {
+    public static String queryResultsTopic(final String ryaInstance, final UUID queryId) {
         requireNonNull(queryId);
-        return "QueryResults-" + queryId.toString();
+        return ryaInstance + "-QueryResults-" + queryId.toString();
+    }
+
+    /**
+     * TODO doc
+     *
+     * @param queryResultsTopic
+     * @return
+     */
+    public static String getRyaInstanceFromQueryResultsTopic(final String queryResultsTopic) {
+        // TODO
+        return "";
+    }
+
+    /**
+     * TODO doc
+     *
+     * @param queryResultsTopic
+     * @return
+     */
+    public static UUID getQueryIdFromQueryResultsTopic(final String queryResultsTopic) {
+        // TODO
+        return null;
     }
 
     /**
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
index 8093951..63d64b9 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
@@ -78,7 +78,7 @@
 
         // Setup the topology that processes the Query.
         final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
-        final String resultsTopic = KafkaTopics.queryResultsTopic(query.getQueryId());
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, query.getQueryId());
 
         try {
             final TopologyBuilder topologyBuilder = topologyFactory.build(query.getSparql(), statementsTopic, resultsTopic, new RandomUUIDFactory());
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
index 529b493..d91edeb 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStream.java
@@ -66,22 +66,22 @@
     }
 
     @Override
-    public QueryResultStream<T> fromStart(final UUID queryId) throws RyaStreamsException {
+    public QueryResultStream<T> fromStart(final String ryaInstance, final UUID queryId) throws RyaStreamsException {
         requireNonNull(queryId);
 
         // Always start at the earliest point within the topic.
-        return makeStream(queryId, "earliest");
+        return makeStream(ryaInstance, queryId, "earliest");
     }
 
     @Override
-    public QueryResultStream<T> fromNow(final UUID queryId) throws RyaStreamsException {
+    public QueryResultStream<T> fromNow(final String ryaInstance, final UUID queryId) throws RyaStreamsException {
         requireNonNull(queryId);
 
         // Always start at the latest point within the topic.
-        return makeStream(queryId, "latest");
+        return makeStream(ryaInstance, queryId, "latest");
     }
 
-    private QueryResultStream<T> makeStream(final UUID queryId, final String autoOffsetResetConfig) {
+    private QueryResultStream<T> makeStream(final String ryaInstance, final UUID queryId, final String autoOffsetResetConfig) {
         // Configure which instance of Kafka to connect to.
         final Properties props = new Properties();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
@@ -106,7 +106,7 @@
         final KafkaConsumer<String, T> consumer = new KafkaConsumer<>(props);
 
         // Register the consumer for the query's results.
-        final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
         consumer.subscribe(Arrays.asList(resultTopic));
 
         // Return the result stream.
diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/KafkaTopicsTest.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/KafkaTopicsTest.java
index a057de7..8eaf080 100644
--- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/KafkaTopicsTest.java
+++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/KafkaTopicsTest.java
@@ -37,7 +37,7 @@
         final String topicName = KafkaTopics.queryChangeLogTopic(ryaInstance);
 
         // Show the rya instance name is able to be extracted from the topic.
-        final Optional<String> resolvedRyaInstance = KafkaTopics.getRyaInstance(topicName);
+        final Optional<String> resolvedRyaInstance = KafkaTopics.getRyaInstanceFromQueryChangeLog(topicName);
         assertEquals(ryaInstance, resolvedRyaInstance.get());
     }
 
@@ -47,7 +47,7 @@
         final String invalidTopic = "thisIsABadTopicName";
 
         // Show there is no Rya Instance name in it.
-        final Optional<String> ryaInstance = KafkaTopics.getRyaInstance(invalidTopic);
+        final Optional<String> ryaInstance = KafkaTopics.getRyaInstanceFromQueryChangeLog(invalidTopic);
         assertFalse( ryaInstance.isPresent() );
     }
 }
\ No newline at end of file
diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
index 8882753..59c08b7 100644
--- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
+++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaGetQueryResultStreamIT.java
@@ -85,6 +85,7 @@
     @Test
     public void fromStart() throws Exception {
         // Create an ID for the query.
+        final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
 
         // Create a list of test VisibilityBindingSets.
@@ -106,7 +107,7 @@
         // Write some entries to the query result topic in Kafka.
         try(final Producer<?, VisibilityBindingSet> producer =
                 KafkaTestUtil.makeProducer(kafka, StringSerializer.class, VisibilityBindingSetSerializer.class)) {
-            final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+            final String resultTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
             for(final VisibilityBindingSet visBs : original) {
                 producer.send(new ProducerRecord<>(resultTopic, visBs));
             }
@@ -115,7 +116,7 @@
         // Use the interactor that is being tested to read all of the visibility binding sets.
         final GetQueryResultStream<VisibilityBindingSet> interactor =
                 new KafkaGetQueryResultStream<>(kafka.getKafkaHostname(), kafka.getKafkaPort(), VisibilityBindingSetDeserializer.class);
-        final List<VisibilityBindingSet> read = pollForResults(500, 3, 3, interactor.fromStart(queryId));
+        final List<VisibilityBindingSet> read = pollForResults(500, 3, 3, interactor.fromStart(ryaInstance, queryId));
 
         // Show the fetched binding sets match the original, as well as their order.
         assertEquals(original, read);
@@ -124,11 +125,12 @@
     @Test
     public void fromNow() throws Exception {
         // Create an ID for the query.
+        final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
 
         try(final Producer<?, VisibilityBindingSet> producer =
                 KafkaTestUtil.makeProducer(kafka, StringSerializer.class, VisibilityBindingSetSerializer.class)) {
-            final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+            final String resultTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
             // Write a single visibility binding set to the query's result topic. This will not appear in the expected results.
             final ValueFactory vf = new ValueFactoryImpl();
@@ -140,7 +142,7 @@
             // Use the interactor that is being tested to read all of the visibility binding sets that appear after this point.
             final GetQueryResultStream<VisibilityBindingSet> interactor =
                     new KafkaGetQueryResultStream<>(kafka.getKafkaHostname(), kafka.getKafkaPort(), VisibilityBindingSetDeserializer.class);
-            try(QueryResultStream<VisibilityBindingSet> results = interactor.fromNow(queryId)) {
+            try(QueryResultStream<VisibilityBindingSet> results = interactor.fromNow(ryaInstance, queryId)) {
                 // Read results from the stream.
                 List<VisibilityBindingSet> read = new ArrayList<>();
                 for(final VisibilityBindingSet visBs : results.poll(500)) {
@@ -178,12 +180,13 @@
     @Test(expected = IllegalStateException.class)
     public void pollClosedStream() throws Exception {
         // Create an ID for the query.
+        final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
 
         // Use the interactor that is being tested to create a result stream and immediately close it.
         final GetQueryResultStream<VisibilityBindingSet> interactor =
                 new KafkaGetQueryResultStream<>(kafka.getKafkaHostname(), kafka.getKafkaPort(), VisibilityBindingSetDeserializer.class);
-        final QueryResultStream<VisibilityBindingSet> results = interactor.fromStart(queryId);
+        final QueryResultStream<VisibilityBindingSet> results = interactor.fromStart(ryaInstance, queryId);
         results.close();
 
         // Try to poll the closed stream.
@@ -193,6 +196,7 @@
     @Test
     public void fromStart_visibilityStatements() throws Exception {
         // Create an ID for the query.
+        final String ryaInstance = UUID.randomUUID().toString();
         final UUID queryId = UUID.randomUUID();
 
         // Create some statements that will be written to the result topic.
@@ -205,7 +209,7 @@
         // Write the entries to the query result topic in Kafka.
         try(final Producer<?, VisibilityStatement> producer =
                 KafkaTestUtil.makeProducer(kafka, StringSerializer.class, VisibilityStatementSerializer.class)) {
-            final String resultTopic = KafkaTopics.queryResultsTopic(queryId);
+            final String resultTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
             for(final VisibilityStatement visStmt : original) {
                 producer.send(new ProducerRecord<>(resultTopic, visStmt));
             }
@@ -214,7 +218,7 @@
         // Use the interactor that is being tested to read all of the visibility binding sets.
         final GetQueryResultStream<VisibilityStatement> interactor =
                 new KafkaGetQueryResultStream<>(kafka.getKafkaHostname(), kafka.getKafkaPort(), VisibilityStatementDeserializer.class);
-        final List<VisibilityStatement> read = pollForResults(500, 3, 3, interactor.fromStart(queryId));
+        final List<VisibilityStatement> read = pollForResults(500, 3, 3, interactor.fromStart(ryaInstance, queryId));
 
         // Show the fetched binding sets match the original, as well as their order.
         assertEquals(original, read);
diff --git a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaRunQueryIT.java b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaRunQueryIT.java
index 5dbd27f..c9abb41 100644
--- a/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaRunQueryIT.java
+++ b/extras/rya.streams/kafka/src/test/java/org/apache/rya/streams/kafka/interactor/KafkaRunQueryIT.java
@@ -90,7 +90,7 @@
         // Add the query to the query repository.
         final StreamsQuery sQuery = queries.add("SELECT * WHERE { ?person <urn:worksAt> ?business . }", true);
         final UUID queryId = sQuery.getQueryId();
-        final String resultsTopic = KafkaTopics.queryResultsTopic(queryId);
+        final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, queryId);
 
         // The thread that will run the tested interactor.
         final Thread testThread = new Thread() {
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
index e746baf..2aa7054 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
@@ -171,11 +171,11 @@
             final Set<String> changeLogTopics = new HashSet<>( listTopicsConsumer.listTopics().keySet() );
 
             // Remove all topics that are not valid Rya Query Change Log topic names.
-            changeLogTopics.removeIf( topic -> !KafkaTopics.getRyaInstance(topic).isPresent() );
+            changeLogTopics.removeIf( topic -> !KafkaTopics.getRyaInstanceFromQueryChangeLog(topic).isPresent() );
 
             // Extract the Rya instance names from the change log topics.
             final Set<String> ryaInstances = changeLogTopics.stream()
-                    .map(topic -> KafkaTopics.getRyaInstance(topic).get() )
+                    .map(topic -> KafkaTopics.getRyaInstanceFromQueryChangeLog(topic).get() )
                     .collect(Collectors.toSet());
 
             // Any Rya instances that are in the old set of topics, but not the new one, have been deleted.
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutor.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutor.java
index 3a59636..4bd022a 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutor.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutor.java
@@ -128,7 +128,7 @@
             // Make sure the Statements topic exists for the query.
             final Set<String> topics = Sets.newHashSet(
                     KafkaTopics.statementsTopic(ryaInstance),
-                    KafkaTopics.queryResultsTopic(query.getQueryId()));
+                    KafkaTopics.queryResultsTopic(ryaInstance, query.getQueryId()));
 
             // Make sure the Query Results topic exists for the query.
             createKafkaTopic.createTopics(topics, 1, 1);
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
index f9c8a03..6358104 100644
--- a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
@@ -138,7 +138,7 @@
             loadStatements.fromCollection(statements);
 
             // Read the output of the streams program.
-            final String resultsTopic = KafkaTopics.queryResultsTopic(sQuery.getQueryId());
+            final String resultsTopic = KafkaTopics.queryResultsTopic(ryaInstance, sQuery.getQueryId());
             resultConsumer.subscribe( Lists.newArrayList(resultsTopic) );
             final List<VisibilityBindingSet> results = KafkaTestUtil.pollForResults(500, 6, 3, resultConsumer);
             assertEquals(expected, results);