@Test(expected = IllegalStateException.class) public void stopQuery_serviceNotStarted() throws Exception { final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class)); executor.stopQuery(UUID.randomUUID()); }
@Override public void stopQuery(final UUID queryId) throws QueryExecutorException { requireNonNull(queryId); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(byQueryId.containsKey(queryId)) { final KafkaStreams streams = byQueryId.get(queryId); streams.close(); final String ryaInstance = ryaInstanceById.remove(queryId); idByRyaInstance.remove(ryaInstance, queryId); byQueryId.remove(queryId); } } finally { lock.unlock(); } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopQuery(final UUID queryId) throws QueryExecutorException { requireNonNull(queryId); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(byQueryId.containsKey(queryId)) { final KafkaStreams streams = byQueryId.get(queryId); streams.close(); final String ryaInstance = ryaInstanceById.remove(queryId); idByRyaInstance.remove(ryaInstance, queryId); byQueryId.remove(queryId); } } finally { lock.unlock(); } } }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopQuery(final UUID queryId) throws QueryExecutorException { requireNonNull(queryId); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(byQueryId.containsKey(queryId)) { final KafkaStreams streams = byQueryId.get(queryId); streams.close(); final String ryaInstance = ryaInstanceById.remove(queryId); idByRyaInstance.remove(ryaInstance, queryId); byQueryId.remove(queryId); } } finally { lock.unlock(); } } LocalQueryExecutor(
            final CreateKafkaTopic createKafkaTopic,
            final KafkaStreamsFactory streamsFactory); }
LocalQueryExecutor extends AbstractIdleService implements QueryExecutor { @Override public void stopQuery(final UUID queryId) throws QueryExecutorException { requireNonNull(queryId); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(byQueryId.containsKey(queryId)) { final KafkaStreams streams = byQueryId.get(queryId); streams.close(); final String ryaInstance = ryaInstanceById.remove(queryId); idByRyaInstance.remove(ryaInstance, queryId); byQueryId.remove(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 stopQuery(final UUID queryId) throws QueryExecutorException { requireNonNull(queryId); checkState(state() == State.RUNNING, "The service must be RUNNING to execute this method."); lock.lock(); try { if(byQueryId.containsKey(queryId)) { final KafkaStreams streams = byQueryId.get(queryId); streams.close(); final String ryaInstance = ryaInstanceById.remove(queryId); idByRyaInstance.remove(ryaInstance, queryId); byQueryId.remove(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; }