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 f9a9458..edcc252 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
@@ -19,9 +19,11 @@
 package org.apache.rya.streams.client.command;
 
 import static java.util.Objects.requireNonNull;
+import static org.apache.rya.streams.kafka.interactor.KafkaTopicPropertiesBuilder.CLEANUP_POLICY_COMPACT;
 
 import java.util.HashSet;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
@@ -33,6 +35,7 @@
 import org.apache.rya.streams.client.RyaStreamsCommand;
 import org.apache.rya.streams.kafka.KafkaTopics;
 import org.apache.rya.streams.kafka.interactor.KafkaRunQuery;
+import org.apache.rya.streams.kafka.interactor.KafkaTopicPropertiesBuilder;
 import org.apache.rya.streams.kafka.queries.KafkaQueryChangeLogFactory;
 import org.apache.rya.streams.kafka.topology.TopologyFactory;
 
@@ -136,7 +139,11 @@
                 final Set<String> topics = new HashSet<>();
                 topics.add( KafkaTopics.statementsTopic(params.ryaInstance) );
                 topics.add( KafkaTopics.queryResultsTopic(params.ryaInstance, queryId) );
-                KafkaTopics.createTopics(params.zookeeperServers, topics, 1, 1);
+
+                final Properties topicProps = new KafkaTopicPropertiesBuilder()
+                    .setCleanupPolicy(CLEANUP_POLICY_COMPACT)
+                    .build();
+                KafkaTopics.createTopics(params.zookeeperServers, topics, 1, 1, Optional.of(topicProps));
 
                 // Run the query that uses those topics.
                 final KafkaRunQuery runQuery = new KafkaRunQuery(
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 f0cc842..e445f47 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
@@ -127,12 +127,14 @@
      * @param topicNames - The topics that will be created. (not null)
      * @param partitions - The number of partitions that each of the topics will have.
      * @param replicationFactor - The replication factor of the topics that are created.
+     * @param topicProperties - The optional properties of the topics to create.
      */
     public static void createTopics(
             final String zookeeperServers,
             final Set<String> topicNames,
             final int partitions,
-            final int replicationFactor) {
+            final int replicationFactor,
+            final Optional<Properties> topicProperties) {
         requireNonNull(zookeeperServers);
         requireNonNull(topicNames);
 
@@ -141,7 +143,7 @@
             zkUtils = ZkUtils.apply(new ZkClient(zookeeperServers, 30000, 30000, ZKStringSerializer$.MODULE$), false);
             for(final String topicName : topicNames) {
                 if(!AdminUtils.topicExists(zkUtils, topicName)) {
-                    AdminUtils.createTopic(zkUtils, topicName, partitions, replicationFactor, new Properties(), RackAwareMode.Disabled$.MODULE$);
+                    AdminUtils.createTopic(zkUtils, topicName, partitions, replicationFactor, topicProperties.orElse(new Properties()), RackAwareMode.Disabled$.MODULE$);
                 }
             }
         }
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/CreateKafkaTopic.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/CreateKafkaTopic.java
index 771e1c8..eb6c9c4 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/CreateKafkaTopic.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/CreateKafkaTopic.java
@@ -20,6 +20,8 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.util.Optional;
+import java.util.Properties;
 import java.util.Set;
 
 import org.apache.rya.streams.kafka.KafkaTopics;
@@ -50,11 +52,13 @@
      * @param topicNames - The topics that will be created. (not null)
      * @param partitions - The number of partitions that each of the topics will have.
      * @param replicationFactor - The replication factor of the topics that are created.
+     * @param topicProperties - The optional properties of the topics to create.
      */
     public void createTopics(
             final Set<String> topicNames,
             final int partitions,
-            final int replicationFactor) {
-        KafkaTopics.createTopics(zookeeperServers, topicNames, partitions, replicationFactor);
+            final int replicationFactor,
+            final Optional<Properties> topicProperties) {
+        KafkaTopics.createTopics(zookeeperServers, topicNames, partitions, replicationFactor, topicProperties);
     }
 }
\ No newline at end of file
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaTopicPropertiesBuilder.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaTopicPropertiesBuilder.java
new file mode 100644
index 0000000..7df9891
--- /dev/null
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/KafkaTopicPropertiesBuilder.java
@@ -0,0 +1,46 @@
+package org.apache.rya.streams.kafka.interactor;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Properties builder to be used when creating new Kafka Topics.
+ *
+ * Descriptions of properties can be found at
+ * {@link https://kafka.apache.org/documentation/#topicconfigs}
+ */
+public class KafkaTopicPropertiesBuilder {
+    /*----- Cleanup Policy -----*/
+    public static final String CLEANUP_POLICY_KEY = "cleanup.policy";
+    public static final String CLEANUP_POLICY_DELETE = "cleanup.policy";
+    public static final String CLEANUP_POLICY_COMPACT = "cleanup.policy";
+
+
+    private Optional<String> cleanupPolicy;
+    /**
+     * Sets the cleanup.policy of the Kafka Topic.
+     *
+     * @param policy - The cleanup policy to use.
+     * @return The builder.
+     */
+    public KafkaTopicPropertiesBuilder setCleanupPolicy(final String policy) {
+        cleanupPolicy = Optional.of(requireNonNull(policy));
+        return this;
+    }
+
+    /**
+     * Builds the Kafka topic properties.
+     * @return The {@link Properties} of the Kafka Topic.
+     */
+    public Properties build() {
+        final Properties props = new Properties();
+
+        if(cleanupPolicy.isPresent()) {
+            props.setProperty(CLEANUP_POLICY_KEY, cleanupPolicy.get());
+        }
+
+        return props;
+    }
+}
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 4bd022a..2341739 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
@@ -24,6 +24,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.locks.ReentrantLock;
@@ -131,7 +132,9 @@
                     KafkaTopics.queryResultsTopic(ryaInstance, query.getQueryId()));
 
             // Make sure the Query Results topic exists for the query.
-            createKafkaTopic.createTopics(topics, 1, 1);
+            // Since this is running in the JVM, the properties are left empty
+            //   so the cleanup.policy will default to delete to reduce memory usage.
+            createKafkaTopic.createTopics(topics, 1, 1, Optional.empty());
 
             // Setup the Kafka Streams job that will execute.
             final KafkaStreams streams = streamsFactory.make(ryaInstance, query);