@Test public void removeQueryFromSingleExecutionQuery() throws IOException { final List<String> result = new LinkedList<>(); final Map<String, String> sourceConf = idAndConfGenerator.generateConf(); final Map<String, String> ocConf = idAndConfGenerator.generateConf(); final Map<String, String> sinkConf = idAndConfGenerator.generateConf(); final TestSource source = generateSource(sourceConf); final PhysicalOperator physicalOp = generateFilterOperator(ocConf, (s) -> true); final PhysicalSink<String> sink = generateSink(sinkConf, result); final ConfigVertex srcVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SOURCE, sourceConf); final ConfigVertex ocVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.OPERATOR, ocConf); final ConfigVertex sinkVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SINK, sinkConf); final Tuple<DAG<ConfigVertex, MISTEdge>, ExecutionDag> dagTuple = generateSimpleDag(source, physicalOp, sink, srcVertex, ocVertex, sinkVertex); final String queryId = "test-query"; srcAndDagMap.put(sourceConf, dagTuple.getValue()); queryIdConfigDagMap.put(queryId, dagTuple.getKey()); configExecutionVertexMap.put(srcVertex, source); configExecutionVertexMap.put(ocVertex, physicalOp); configExecutionVertexMap.put(sinkVertex, sink); executionDags.add(dagTuple.getValue()); executionVertexCountMap.put(source, 1); executionVertexCountMap.put(physicalOp, 1); executionVertexCountMap.put(sink, 1); executionVertexDagMap.put(source, dagTuple.getValue()); executionVertexDagMap.put(physicalOp, dagTuple.getValue()); executionVertexDagMap.put(sink, dagTuple.getValue()); queryRemover.deleteQuery(queryId); Assert.assertEquals(0, srcAndDagMap.size()); Assert.assertNull(queryIdConfigDagMap.get(queryId)); Assert.assertNull(configExecutionVertexMap.get(srcVertex)); Assert.assertNull(configExecutionVertexMap.get(ocVertex)); Assert.assertNull(configExecutionVertexMap.get(sinkVertex)); Assert.assertEquals(0, executionDags.values().size()); Assert.assertNull(executionVertexCountMap.get(source)); Assert.assertNull(executionVertexCountMap.get(physicalOp)); Assert.assertNull(executionVertexCountMap.get(sink)); Assert.assertNull(executionVertexDagMap.get(source)); Assert.assertNull(executionVertexDagMap.get(physicalOp)); Assert.assertNull(executionVertexDagMap.get(sink)); try { source.send("data"); Assert.fail("Source should be closed"); } catch (final InvalidDataSendException e) { } }
@Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void deleteQuery(final String queryId); @Override void deleteAllQueries(); }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void deleteQuery(final String queryId); @Override void deleteAllQueries();  }