diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml b/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
index 01da2dc..fd624eb 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
@@ -1,25 +1,16 @@
 <?xml version="1.0" encoding="utf-8"?>
-<!--
-
-    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
-    regarding copyright ownership.  The ASF licenses this file
-    to you under the Apache License, Version 2.0 (the
-    "License"); you may not use this file except in compliance
-    with the License.  You may obtain a copy of the License at
-
-        http://www.apache.org/licenses/LICENSE-2.0
-
-    Unless required by applicable law or agreed to in writing,
-    software distributed under the License is distributed on an
-    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-    KIND, either express or implied.  See the License for the
-    specific language governing permissions and limitations
-    under the License.
-
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<!-- 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 regarding copyright ownership. The ASF licenses this file to 
+    you under the Apache License, Version 2.0 (the "License"); you may not use 
+    this file except in compliance with the License. You may obtain a copy of 
+    the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required 
+    by applicable law or agreed to in writing, software distributed under the 
+    License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS 
+    OF ANY KIND, either express or implied. See the License for the specific 
+    language governing permissions and limitations under the License. -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 
     <parent>
         <groupId>org.apache.rya</groupId>
@@ -41,12 +32,14 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
-
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <!-- Uncommment this block when rya.pcj.fluo.app becomes a leaf project.  RYA-341 -->
-            <!-- <version>13.0</version>  Overriding Rya's Guava version to be compatible with Fluo's required version.  Alternative is relocation with shade. -->
+            <!-- Uncommment this block when rya.pcj.fluo.app becomes a leaf 
+                project. RYA-341 -->
+            <!-- <version>13.0</version> Overriding Rya's Guava version to 
+                be compatible with Fluo's required version. Alternative is relocation with 
+                shade. -->
         </dependency>
 
         <!-- Rya Runtime Dependencies. -->
@@ -75,7 +68,7 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
-        
+
         <dependency>
             <groupId>com.esotericsoftware</groupId>
             <artifactId>kryo</artifactId>
@@ -87,6 +80,11 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
index c0cfa1d..9e47132 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
@@ -36,8 +36,9 @@
 import org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata;
 import org.apache.rya.indexing.pcj.fluo.app.query.ConstructQueryMetadata;
 import org.apache.rya.indexing.pcj.fluo.app.query.FilterMetadata;
-import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataCache;
 import org.apache.rya.indexing.pcj.fluo.app.query.JoinMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.query.MetadataCacheSupplier;
 import org.apache.rya.indexing.pcj.fluo.app.query.PeriodicQueryMetadata;
 import org.apache.rya.indexing.pcj.fluo.app.query.ProjectionMetadata;
 import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
@@ -55,7 +56,7 @@
 public abstract class BindingSetUpdater extends AbstractObserver {
     private static final Logger log = Logger.getLogger(BindingSetUpdater.class);
     // DAO
-    protected final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
+    protected final FluoQueryMetadataCache queryDao = MetadataCacheSupplier.getOrCreateCache();
 
     // Updaters
     private final JoinResultUpdater joinUpdater = new JoinResultUpdater();
@@ -117,9 +118,9 @@
                 } catch (final Exception e) {
                     throw new RuntimeException("Could not process a Query node.", e);
                 }
-                break;    
-                
-            case CONSTRUCT: 
+                break;
+
+            case CONSTRUCT:
                 final ConstructQueryMetadata constructQuery = queryDao.readConstructQueryMetadata(tx, parentNodeId);
                 try{
                     constructUpdater.updateConstructQueryResults(tx, observedBindingSet, constructQuery);
@@ -127,7 +128,7 @@
                     throw new RuntimeException("Could not process a Query node.", e);
                 }
                 break;
-                
+
             case FILTER:
                 final FilterMetadata parentFilter = queryDao.readFilterMetadata(tx, parentNodeId);
                 try {
@@ -145,7 +146,7 @@
                     throw new RuntimeException("Could not process a Join node.", e);
                 }
                 break;
-                
+
             case PERIODIC_QUERY:
                 final PeriodicQueryMetadata parentPeriodicQuery = queryDao.readPeriodicQueryMetadata(tx, parentNodeId);
                 try{
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/ConstructQueryResultObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/ConstructQueryResultObserver.java
index 61e7244..09d9ede 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/ConstructQueryResultObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/ConstructQueryResultObserver.java
@@ -29,6 +29,8 @@
 import org.apache.rya.indexing.pcj.fluo.app.NodeType;
 import org.apache.rya.indexing.pcj.fluo.app.export.IncrementalRyaSubGraphExporter;
 import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataCache;
+import org.apache.rya.indexing.pcj.fluo.app.query.MetadataCacheSupplier;
 
 /**
  * Monitors the Column {@link FluoQueryColumns#CONSTRUCT_STATEMENTS} for new
@@ -40,6 +42,7 @@
 public class ConstructQueryResultObserver extends AbstractObserver {
 
     private static final Logger log = Logger.getLogger(ConstructQueryResultObserver.class);
+    protected final FluoQueryMetadataCache queryDao = MetadataCacheSupplier.getOrCreateCache();
 
     @Override
     public ObservedColumn getObservedColumn() {
@@ -48,14 +51,14 @@
 
     @Override
     public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
-        
+
         //Build row for parent that result will be written to
         BindingSetRow bsRow = BindingSetRow.make(row);
         String constructNodeId = bsRow.getNodeId();
         String bsString= bsRow.getBindingSetString();
-        String parentNodeId = tx.get(Bytes.of(constructNodeId), FluoQueryColumns.CONSTRUCT_PARENT_NODE_ID).toString();
+        String parentNodeId = queryDao.readMetadadataEntry(tx, constructNodeId, FluoQueryColumns.CONSTRUCT_PARENT_NODE_ID).toString();
         String rowString = parentNodeId + IncrementalUpdateConstants.NODEID_BS_DELIM + bsString;
-        
+
         //Get NodeType of the parent node
         NodeType parentType = NodeType.fromNodeId(parentNodeId).get();
         //Get data for the ConstructQuery result
@@ -63,5 +66,5 @@
         //Write result to parent
         tx.set(Bytes.of(rowString), parentType.getResultColumn(), bytes);
     }
-   
+
 }
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
index 9514932..78d0ec5 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
@@ -35,7 +35,8 @@
 import org.apache.rya.indexing.pcj.fluo.app.export.rya.PeriodicBindingSetExporterFactory;
 import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaBindingSetExporterFactory;
 import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaSubGraphExporterFactory;
-import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataCache;
+import org.apache.rya.indexing.pcj.fluo.app.query.MetadataCacheSupplier;
 import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -49,8 +50,7 @@
 public class QueryResultObserver extends AbstractObserver {
 
     private static final Logger log = LoggerFactory.getLogger(QueryResultObserver.class);
-    private static final FluoQueryMetadataDAO DAO = new FluoQueryMetadataDAO();
-
+    protected final FluoQueryMetadataCache queryDao = MetadataCacheSupplier.getOrCreateCache();
     /**
      * Builders for each type of {@link IncrementalBindingSetExporter} we support.
      */
@@ -101,7 +101,7 @@
 
         // Read the queryId from the row and get the QueryMetadata.
         final String queryId = row.split(NODEID_BS_DELIM)[0];
-        final QueryMetadata metadata = DAO.readQueryMetadata(tx, queryId);
+        final QueryMetadata metadata = queryDao.readQueryMetadata(tx, queryId);
 
         // Read the Child Binding Set that will be exported.
         final Bytes valueBytes = tx.get(brow, col);
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
index 2d7f390..d6fd8bd 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
@@ -34,7 +34,8 @@
 import org.apache.rya.api.domain.RyaStatement;
 import org.apache.rya.indexing.pcj.fluo.app.IncUpdateDAO;
 import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
-import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataCache;
+import org.apache.rya.indexing.pcj.fluo.app.query.MetadataCacheSupplier;
 import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
 import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
 import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
@@ -55,7 +56,7 @@
     private static final Logger log = LoggerFactory.getLogger(TripleObserver.class);
 
     private static final VisibilityBindingSetSerDe BS_SERDE = new VisibilityBindingSetSerDe();
-    private static final FluoQueryMetadataDAO QUERY_METADATA_DAO = new FluoQueryMetadataDAO();
+    private static final FluoQueryMetadataCache QUERY_METADATA_DAO = MetadataCacheSupplier.getOrCreateCache();
     private static final VisibilityBindingSetStringConverter VIS_BS_CONVERTER = new VisibilityBindingSetStringConverter();
 
     public TripleObserver() {}
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCache.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCache.java
new file mode 100644
index 0000000..8adc40d
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCache.java
@@ -0,0 +1,241 @@
+/*
+ * 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
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */package org.apache.rya.indexing.pcj.fluo.app.query;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.util.concurrent.Callable;
+
+import org.apache.fluo.api.client.SnapshotBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.rya.indexing.pcj.fluo.app.NodeType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
+/**
+ * Wrapper for {@link FluoQueryMetadataDAO} that caches any metadata that has been retrieved from Fluo. This class first
+ * checks the cache to see if the metadata is present before delegating to the underlying DAO method to retrieve the
+ * data.
+ *
+ */
+public class FluoQueryMetadataCache extends FluoQueryMetadataDAO {
+
+    private static final Logger LOG = LoggerFactory.getLogger(FluoQueryMetadataCache.class);
+    private final FluoQueryMetadataDAO dao;
+    private final Cache<String, CommonNodeMetadata> commonNodeMetadataCache;
+    private final Cache<String, Bytes> metadataCache;
+    private int capacity;
+    private int concurrencyLevel;
+
+    /**
+     * Creates a FluoQueryMetadataCache with the specified capacity. Old, unused results are evicted as necessary.
+     *
+     * @param capacity - max size of the cache
+     */
+    public FluoQueryMetadataCache(FluoQueryMetadataDAO dao, int capacity, int concurrencyLevel) {
+        this.dao = dao;
+        commonNodeMetadataCache = CacheBuilder.newBuilder().concurrencyLevel(concurrencyLevel).maximumSize(capacity).build();
+        metadataCache = CacheBuilder.newBuilder().concurrencyLevel(concurrencyLevel).maximumSize(capacity).build();
+        this.capacity = capacity;
+        this.concurrencyLevel = concurrencyLevel;
+    }
+
+    /**
+     * @return - capacity of this cache in terms of max number of entries
+     */
+    public int getCapacity() {
+        return capacity;
+    }
+
+    /**
+     * @return - concurrencyLevel of this cache,in terms of number of partitions that distinct threads can operate on
+     *         without waiting for other threads
+     */
+    public int getConcurrencyLevel() {
+        return concurrencyLevel;
+    }
+
+    @Override
+    public StatementPatternMetadata readStatementPatternMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.STATEMENT_PATTERN);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (StatementPatternMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}", nodeId);
+                    return dao.readStatementPatternMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access StatementPatternMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public JoinMetadata readJoinMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.JOIN);
+            LOG.debug("Retrieving Metadata from Cache: {}.", nodeId);
+            return (JoinMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readJoinMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access JoinMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public FilterMetadata readFilterMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.FILTER);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (FilterMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readFilterMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access FilterMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public ProjectionMetadata readProjectionMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        checkArgument(type.isPresent() && type.get() == NodeType.PROJECTION);
+        LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+        try {
+            return (ProjectionMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readProjectionMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access ProjectionMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public AggregationMetadata readAggregationMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.AGGREGATION);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (AggregationMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readAggregationMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access AggregationMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public ConstructQueryMetadata readConstructQueryMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.CONSTRUCT);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (ConstructQueryMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readConstructQueryMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access ConstructQueryMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public PeriodicQueryMetadata readPeriodicQueryMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.PERIODIC_QUERY);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (PeriodicQueryMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readPeriodicQueryMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access PeriodicQueryMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    @Override
+    public QueryMetadata readQueryMetadata(SnapshotBase tx, String nodeId) {
+        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
+        try {
+            checkArgument(type.isPresent() && type.get() == NodeType.QUERY);
+            LOG.debug("Retrieving Metadata from Cache: {}", nodeId);
+            return (QueryMetadata) commonNodeMetadataCache.get(nodeId, new Callable<CommonNodeMetadata>() {
+                @Override
+                public CommonNodeMetadata call() throws Exception {
+                    LOG.debug("Seeking Metadata from Fluo Table: {}.", nodeId);
+                    return dao.readQueryMetadata(tx, nodeId);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access QueryMetadata for nodeId: " + nodeId, e);
+        }
+    }
+
+    public Bytes readMetadadataEntry(SnapshotBase tx, String rowId, Column column) {
+        Optional<NodeType> type = NodeType.fromNodeId(rowId);
+        try {
+            checkArgument(type.isPresent() && type.get().getMetaDataColumns().contains(column));
+            return metadataCache.get(getKey(rowId, column), new Callable<Bytes>() {
+                @Override
+                public Bytes call() throws Exception {
+                    return tx.get(Bytes.of(rowId), column);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to access Metadata Entry with rowId: " + rowId + " and column: " + column, e);
+        }
+    }
+
+    private String getKey(String row, Column column) {
+        return row + ":" + column.getsQualifier() + ":" + column.getsQualifier();
+    }
+}
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/MetadataCacheSupplier.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/MetadataCacheSupplier.java
new file mode 100644
index 0000000..faab952
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/MetadataCacheSupplier.java
@@ -0,0 +1,44 @@
+package org.apache.rya.indexing.pcj.fluo.app.query;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetadataCacheSupplier {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MetadataCacheSupplier.class);
+    private static FluoQueryMetadataCache CACHE;
+    private static boolean initialized = false;
+    private static final int DEFAULT_CAPACITY = 10000;
+    private static final int DEFAULT_CONCURRENCY = 8;
+
+    /**
+     * Returns an existing cache with the specified instance name, or creates a cache. The created cache will have the
+     * indicated capacity and concurrencyLevel if one is provided.
+     *
+     * @param capacity - capacity used to create a new cache
+     * @param concurrencyLevel - concurrencyLevel used to create a new cache
+     */
+    public static FluoQueryMetadataCache getOrCreateCache(int capacity, int concurrencyLevel) {
+        if (!initialized) {
+            LOG.debug("Cache has not been initialized.  Initializing cache with capacity: {} and concurrencylevel: {}", capacity,
+                    concurrencyLevel);
+            CACHE = new FluoQueryMetadataCache(new FluoQueryMetadataDAO(), capacity, concurrencyLevel);
+            initialized = true;
+        } else {
+            LOG.debug("Cache has already been initialized.  Returning cache with capacity: {} and concurrencylevel: {}",
+                    CACHE.getCapacity(), CACHE.getConcurrencyLevel());
+        }
+        return CACHE;
+    }
+
+    /**
+     * Returns cache with the name {@link FluoQueryMetadataCache#FLUO_CACHE_INSTANCE} if it exists, otherwise creates it
+     * with a default size of 10000 entries and a default concurrency level of 8.
+     *
+     * @return - FluoQueryMetadataCache with default instance name and default capacity and concurrency
+     */
+    public static FluoQueryMetadataCache getOrCreateCache() {
+        return getOrCreateCache(DEFAULT_CAPACITY, DEFAULT_CONCURRENCY);
+    }
+
+}
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCacheTest.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCacheTest.java
new file mode 100644
index 0000000..3df3708
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataCacheTest.java
@@ -0,0 +1,34 @@
+package org.apache.rya.indexing.pcj.fluo.app.query;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
+
+import org.apache.fluo.api.client.Transaction;
+import org.apache.rya.indexing.pcj.fluo.app.NodeType;
+import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class FluoQueryMetadataCacheTest {
+
+    @Test
+    public void testCache() {
+        FluoQueryMetadataDAO mockDAO = Mockito.mock(FluoQueryMetadataDAO.class);
+        Transaction mockTx = Mockito.mock(Transaction.class);
+        String nodeId = NodeType.generateNewFluoIdForType(NodeType.STATEMENT_PATTERN);
+        StatementPatternMetadata metadata = StatementPatternMetadata.builder(nodeId).setParentNodeId("parent")
+                .setStatementPattern("pattern").setVarOrder(new VariableOrder("xyz")).build();
+        when(mockDAO.readStatementPatternMetadata(mockTx, nodeId)).thenReturn(metadata);
+
+        FluoQueryMetadataCache cache = new FluoQueryMetadataCache(mockDAO, 20, 2);
+
+        assertEquals(metadata, cache.readStatementPatternMetadata(mockTx, nodeId));
+
+        cache.readStatementPatternMetadata(mockTx, nodeId);
+        cache.readStatementPatternMetadata(mockTx, nodeId);
+        cache.readStatementPatternMetadata(mockTx, nodeId);
+        cache.readStatementPatternMetadata(mockTx, nodeId);
+
+        Mockito.verify(mockDAO, Mockito.times(1)).readStatementPatternMetadata(mockTx, nodeId);
+    }
+}
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/FluoLatencyIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/FluoLatencyIT.java
new file mode 100644
index 0000000..fabf512
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/FluoLatencyIT.java
@@ -0,0 +1,169 @@
+package org.apache.rya.indexing.pcj.fluo.integration;
+
+import static java.util.Objects.requireNonNull;
+
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import javax.xml.datatype.DatatypeConfigurationException;
+import javax.xml.datatype.DatatypeFactory;
+
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.scanner.CellScanner;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.rya.api.client.CreatePCJ.ExportStrategy;
+import org.apache.rya.indexing.pcj.fluo.api.CreateFluoPcj;
+import org.apache.rya.indexing.pcj.fluo.app.util.FluoQueryUtils;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.apache.rya.pcj.fluo.test.base.KafkaExportITBase;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.openrdf.model.Literal;
+import org.openrdf.model.Statement;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.repository.sail.SailRepositoryConnection;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+public class FluoLatencyIT extends KafkaExportITBase {
+    private static ValueFactory vf;
+    private static DatatypeFactory dtf;
+
+    @BeforeClass
+    public static void init() throws DatatypeConfigurationException {
+        vf = new ValueFactoryImpl();
+        dtf = DatatypeFactory.newInstance();
+    }
+
+    @Test
+    public void resultsExported() throws Exception {
+
+        final String sparql = "prefix time: <http://www.w3.org/2006/time#> " + "select ?type (count(?obs) as ?total) where { "
+                + "    ?obs <uri:hasTime> ?time. " + "    ?obs <uri:hasObsType> ?type " + "} " + "group by ?type";
+
+//        final String sparql = "prefix time: <http://www.w3.org/2006/time#> " + "select ?type ?obs where { "
+//                + "    ?obs <uri:hasTime> ?time. " + "    ?obs <uri:hasObsType> ?type " + "}";
+
+        try (FluoClient fluoClient = FluoFactory.newClient(super.getFluoConfiguration())) {
+            // Tell the Fluo app to maintain the PCJ.
+            String pcjId = FluoQueryUtils.createNewPcjId();
+            FluoConfiguration conf = super.getFluoConfiguration();
+            new CreateFluoPcj().createPcj(pcjId, sparql, Sets.newHashSet(ExportStrategy.KAFKA), fluoClient);
+            SailRepositoryConnection conn = super.getRyaSailRepository().getConnection();
+
+            long start = System.currentTimeMillis();
+            int numReturned = 0;
+            int numObs = 10;
+            int numTypes = 5;
+            int numExpected = 0;
+            int increment = numObs*numTypes;
+            while (System.currentTimeMillis() - start < 60000) {
+                List<Statement> statements = generate(10, 5, "car_", numExpected, ZonedDateTime.now());
+                conn.add(statements);
+                numExpected += increment;
+                System.out.println("Num Accumulo Entries: " + getNumAccEntries(conf.getAccumuloTable()) + " Num Fluo Entries: "
+                        + getNumFluoEntries(fluoClient));
+                numReturned += readAllResults(pcjId).size();
+                System.out
+                        .println("Expected: " + numExpected + " NumReturned: " + numReturned + " Difference: " + (numExpected - numReturned));
+//                FluoITHelper.printFluoTable(conf);
+                Thread.sleep(30000);
+            }
+        }
+    }
+
+    /**
+     * Generates (numObservationsPerType x numTypes) statements of the form:
+     *
+     * <pre>
+     * urn:obs_n uri:hasTime zonedTime
+     * urn:obs_n uri:hasObsType typePrefix_m
+     * </pre>
+     *
+     * Where the n in urn:obs_n is the ith value in 0 to (numObservationsPerType x numTypes) with an offset specified by
+     * observationOffset, and where m in typePrefix_m is the jth value in 0 to numTypes.
+     *
+     * @param numObservationsPerType - The quantity of observations per type to generate.
+     * @param numTypes - The number of types to generate observations for.
+     * @param typePrefix - The prefix to be used for the type literal in the statement.
+     * @param observationOffset - The offset to be used for determining the value of n in the above statements.
+     * @param zonedTime - The time to be used for all observations generated.
+     * @return A new list of all generated Statements.
+     */
+    public List<Statement> generate(final long numObservationsPerType, final int numTypes, final String typePrefix,
+            final long observationOffset, final ZonedDateTime zonedTime) {
+        final String time = zonedTime.format(DateTimeFormatter.ISO_INSTANT);
+        final Literal litTime = vf.createLiteral(dtf.newXMLGregorianCalendar(time));
+        final List<Statement> statements = Lists.newArrayList();
+
+        for (long i = 0; i < numObservationsPerType; i++) {
+            for (int j = 0; j < numTypes; j++) {
+                final long observationId = observationOffset + i * numTypes + j;
+                // final String obsId = "urn:obs_" + Long.toHexString(observationId) + "_" + observationId;
+                // final String obsId = "urn:obs_" + observationId;
+                final String obsId = "urn:obs_" + String.format("%020d", observationId);
+                final String type = typePrefix + j;
+                // logger.info(obsId + " " + type + " " + litTime);
+                statements.add(vf.createStatement(vf.createURI(obsId), vf.createURI("uri:hasTime"), litTime));
+                statements.add(vf.createStatement(vf.createURI(obsId), vf.createURI("uri:hasObsType"), vf.createLiteral(type)));
+            }
+        }
+
+        return statements;
+    }
+
+    private Set<VisibilityBindingSet> readAllResults(final String pcjId) throws Exception {
+        requireNonNull(pcjId);
+
+        // Read all of the results from the Kafka topic.
+        final Set<VisibilityBindingSet> results = new HashSet<>();
+
+        try (final KafkaConsumer<String, VisibilityBindingSet> consumer = makeConsumer(pcjId)) {
+            final ConsumerRecords<String, VisibilityBindingSet> records = consumer.poll(5000);
+            final Iterator<ConsumerRecord<String, VisibilityBindingSet>> recordIterator = records.iterator();
+            while (recordIterator.hasNext()) {
+                results.add(recordIterator.next().value());
+            }
+        }
+
+        return results;
+    }
+
+    private int getNumAccEntries(String tableName) throws TableNotFoundException {
+        Scanner scanner = super.getAccumuloConnector().createScanner(tableName, new Authorizations());
+        int count = 0;
+        for (Map.Entry<Key, Value> entry : scanner) {
+            count++;
+        }
+        return count;
+    }
+
+    private int getNumFluoEntries(FluoClient client) {
+        Transaction tx = client.newTransaction();
+        CellScanner scanner = tx.scanner().build();
+        int count = 0;
+        for (RowColumnValue rcv : scanner) {
+            count++;
+        }
+        return count;
+    }
+
+}
diff --git a/extras/rya.pcj.fluo/pcj.fluo.test.base/src/main/java/org/apache/rya/pcj/fluo/test/base/KafkaExportITBase.java b/extras/rya.pcj.fluo/pcj.fluo.test.base/src/main/java/org/apache/rya/pcj/fluo/test/base/KafkaExportITBase.java
index 59fe54f..7b16dcf 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.test.base/src/main/java/org/apache/rya/pcj/fluo/test/base/KafkaExportITBase.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.test.base/src/main/java/org/apache/rya/pcj/fluo/test/base/KafkaExportITBase.java
@@ -69,7 +69,6 @@
 import org.apache.rya.sail.config.RyaSailFactory;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Test;
 import org.openrdf.model.Statement;
 import org.openrdf.repository.sail.SailRepositoryConnection;
 import org.openrdf.sail.Sail;
@@ -129,7 +128,7 @@
         final KafkaBindingSetExporterParameters kafkaParams = new KafkaBindingSetExporterParameters(exportParams);
         kafkaParams.setUseKafkaBindingSetExporter(true);
         kafkaParams.setKafkaBootStrapServers(BROKERHOST + ":" + BROKERPORT);
-        
+
         final KafkaSubGraphExporterParameters kafkaConstructParams = new KafkaSubGraphExporterParameters(exportParams);
         kafkaConstructParams.setUseKafkaSubgraphExporter(true);
 
@@ -262,7 +261,7 @@
      * If this test fails then its a testing environment issue, not with Rya.
      * Source: https://github.com/asmaier/mini-kafka
      */
-    @Test
+//    @Test
     public void embeddedKafkaTest() throws Exception {
         // create topic
         final String topic = "testTopic";
@@ -339,9 +338,9 @@
         // The PCJ Id is the topic name the results will be written to.
         return pcjId;
     }
-    
+
     protected void loadData(final Collection<Statement> statements) throws Exception {
-        
+
         requireNonNull(statements);
 
         final SailRepositoryConnection ryaConn = getRyaSailRepository().getConnection();
@@ -352,7 +351,7 @@
 
         // Wait for the Fluo application to finish computing the end result.
         super.getMiniFluo().waitForObservers();
-        
+
     }
 
 }