@Test public void stopAll() throws Exception { final String ryaInstance1 = "rya1"; final StreamsQuery query1= new StreamsQuery(UUID.randomUUID(), "SELECT * WHERE { ?a ?b ?c. }", true, false); final String ryaInstance2 = "rya2"; final StreamsQuery query2= new StreamsQuery(UUID.randomUUID(), "SELECT * WHERE { ?a ?b ?c. }", true, false); final KafkaStreamsFactory jobFactory = mock(KafkaStreamsFactory.class); final KafkaStreams queryJob1 = mock(KafkaStreams.class); when(jobFactory.make(eq(ryaInstance1), eq(query1))).thenReturn(queryJob1); final KafkaStreams queryJob2 = mock(KafkaStreams.class); when(jobFactory.make(eq(ryaInstance2), eq(query2))).thenReturn(queryJob2); final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory); executor.startAndWait(); try { executor.startQuery(ryaInstance1, query1); executor.startQuery(ryaInstance2, query2); verify(queryJob1).start(); verify(queryJob2).start(); executor.stopAll(ryaInstance2); verify(queryJob1, never()).close(); verify(queryJob2).close(); } finally { executor.stopAndWait(); } }
@Override public void stopAll(final String ryaInstanceName) throws QueryExecutorException { requireNonNull(ryaInstanceName); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(idByRyaInstance.containsKey(ryaInstanceName)) { final Set<UUID> queryIds = new HashSet<>( idByRyaInstance.get(ryaInstanceName) ); for(final UUID queryId : queryIds) { stopQuery(queryId); } } } finally { lock.unlock(); } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopAll(final String ryaInstanceName) throws QueryExecutorException { requireNonNull(ryaInstanceName); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(idByRyaInstance.containsKey(ryaInstanceName)) { final Set<UUID> queryIds = new HashSet<>( idByRyaInstance.get(ryaInstanceName) ); for(final UUID queryId : queryIds) { stopQuery(queryId); } } } finally { lock.unlock(); } } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopAll(final String ryaInstanceName) throws QueryExecutorException { requireNonNull(ryaInstanceName); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(idByRyaInstance.containsKey(ryaInstanceName)) { final Set<UUID> queryIds = new HashSet<>( idByRyaInstance.get(ryaInstanceName) ); for(final UUID queryId : queryIds) { stopQuery(queryId); } } } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopAll(final String ryaInstanceName) throws QueryExecutorException { requireNonNull(ryaInstanceName); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(idByRyaInstance.containsKey(ryaInstanceName)) { final Set<UUID> queryIds = new HashSet<>( idByRyaInstance.get(ryaInstanceName) ); for(final UUID queryId : queryIds) { stopQuery(queryId); } } } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); @Override void startQuery(final String ryaInstance, final StreamsQuery query); @Override void stopQuery(final UUID queryId); @Override void stopAll(final String ryaInstanceName); @Override Set<UUID> getRunningQueryIds(); }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopAll(final String ryaInstanceName) throws QueryExecutorException { requireNonNull(ryaInstanceName); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(idByRyaInstance.containsKey(ryaInstanceName)) { final Set<UUID> queryIds = new HashSet<>( idByRyaInstance.get(ryaInstanceName) ); for(final UUID queryId : queryIds) { stopQuery(queryId); } } } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); @Override void startQuery(final String ryaInstance, final StreamsQuery query); @Override void stopQuery(final UUID queryId); @Override void stopAll(final String ryaInstanceName); @Override Set<UUID> getRunningQueryIds(); static ReentrantLock lock; }