RYA-451 Fixing threading issues with the QueryManager class.
diff --git a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java
index 7194834..11423bd 100644
--- a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java
+++ b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java
@@ -85,7 +85,7 @@
}
return false;
}
-
+
@Override
public String toString() {
final StringBuilder sb = new StringBuilder();
@@ -95,7 +95,7 @@
sb.append(getSparql() + "\n");
sb.append("Is ");
if (!isActive) {
- sb.append(" Not ");
+ sb.append("Not ");
}
sb.append("Running.\n");
return sb.toString();
diff --git a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java
index 5fb0297..95c1922 100644
--- a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java
+++ b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java
@@ -51,9 +51,9 @@
*/
@DefaultAnnotation(NonNull.class)
public class InMemoryQueryRepository extends AbstractScheduledService implements QueryRepository {
- private static final Logger LOG = LoggerFactory.getLogger(InMemoryQueryRepository.class);
+ private static final Logger log = LoggerFactory.getLogger(InMemoryQueryRepository.class);
- private final ReentrantLock lock = new ReentrantLock();
+ private final ReentrantLock lock = new ReentrantLock(true);
/**
* The change log that is the ground truth for describing what the queries look like.
@@ -198,7 +198,6 @@
@Override
protected void shutDown() throws Exception {
- super.shutDown();
lock.lock();
try {
changeLog.close();
@@ -211,11 +210,12 @@
* Updates the {@link #queriesCache} to reflect the latest position within the {@link #changeLog}.
*/
private void updateCache() {
- requireNonNull(changeLog);
+ log.trace("updateCache() - Enter");
CloseableIteration<ChangeLogEntry<QueryChange>, QueryChangeLogException> it = null;
try {
// Iterate over everything since the last position that was handled within the change log.
+ log.debug("Starting cache position:" + cachePosition);
if(cachePosition.isPresent()) {
it = changeLog.readFromPosition(cachePosition.get() + 1);
} else {
@@ -228,6 +228,8 @@
final QueryChange change = entry.getEntry();
final UUID queryId = change.getQueryId();
+ log.debug("Updating the cache to reflect:\n" + change);
+
switch(change.getChangeType()) {
case CREATE:
final StreamsQuery query = new StreamsQuery(
@@ -253,15 +255,17 @@
break;
}
+ log.debug("Notifying listeners with the updated state.");
final Optional<StreamsQuery> newQueryState = Optional.ofNullable(queriesCache.get(queryId));
listeners.forEach(listener -> listener.notify(entry, newQueryState));
cachePosition = Optional.of( entry.getPosition() );
+ log.debug("New chache position: " + cachePosition);
}
} catch (final QueryChangeLogException e) {
// Rethrow the exception because the object the supplier tried to create could not be created.
- throw new RuntimeException("Could not initialize the " + InMemoryQueryRepository.class.getName(), e);
+ throw new RuntimeException("Could not update the cache of " + InMemoryQueryRepository.class.getName(), e);
} finally {
// Try to close the iteration if it was opened.
@@ -270,18 +274,22 @@
it.close();
}
} catch (final QueryChangeLogException e) {
- LOG.error("Could not close the " + CloseableIteration.class.getName(), e);
+ log.error("Could not close the " + CloseableIteration.class.getName(), e);
}
+
+ log.trace("updateCache() - Exit");
}
}
@Override
protected void runOneIteration() throws Exception {
+ log.trace("runOneIteration() - Enter");
lock.lock();
try {
updateCache();
} finally {
lock.unlock();
+ log.trace("runOneIteration() - Exit");
}
}
@@ -292,17 +300,25 @@
@Override
public Set<StreamsQuery> subscribe(final QueryChangeLogListener listener) {
+ log.trace("subscribe(listener) - Enter");
+
//locks to prevent the current state from changing while subscribing.
lock.lock();
+ log.trace("subscribe(listener) - Acquired lock");
try {
listeners.add(listener);
+ log.trace("subscribe(listener) - Listener Registered");
//return the current state of the query repository
- return queriesCache.values()
+ final Set<StreamsQuery> queries = queriesCache.values()
.stream()
.collect(Collectors.toSet());
+ log.trace("subscribe(listener) - Returning " + queries.size() + " existing queries");
+ return queries;
} finally {
+ log.trace("subscribe(listener) - Releasing lock");
lock.unlock();
+ log.trace("subscribe(listener) - Exit");
}
}
diff --git a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java
index d283957..d34a394 100644
--- a/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java
+++ b/extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java
@@ -110,6 +110,16 @@
return false;
}
+ @Override
+ public String toString() {
+ return "QueryChange: {" +
+ " Query ID: " + queryId + ",\n" +
+ " Change Type: " + changeType + ",\n" +
+ " Is Active: " + isActive + ",\n" +
+ " SPARQL: " + sparql + "\n" +
+ "}";
+ }
+
/**
* Create a {@link QueryChange} that represents a new SPARQL query that will be managed by Rya Streams.
*
diff --git a/extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java b/extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java
index 3b3d48a..d7e116b 100644
--- a/extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java
+++ b/extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java
@@ -93,13 +93,9 @@
// Create a new totally in memory QueryRepository.
final QueryRepository initializedQueries = new InMemoryQueryRepository( changeLog, SCHEDULE );
- try {
- // Listing the queries should work using an initialized change log.
- final Set<StreamsQuery> stored = initializedQueries.list();
- assertEquals(expected, stored);
- } finally {
- queries.stop();
- }
+ // Listing the queries should work using an initialized change log.
+ final Set<StreamsQuery> stored = initializedQueries.list();
+ assertEquals(expected, stored);
} finally {
queries.stop();
}
@@ -166,7 +162,7 @@
final StreamsQuery query = queries.add("query 1", true);
final Set<StreamsQuery> existing = queries.subscribe((queryChangeEvent, newQueryState) -> {
- final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<QueryChange>(1L,
+ final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<>(1L,
QueryChange.create(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
final Optional<StreamsQuery> expectedQueryState = Optional.of(
new StreamsQuery(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
@@ -197,7 +193,7 @@
//show listener on repo that query was added to is being notified of the new query.
final CountDownLatch repo1Latch = new CountDownLatch(1);
queries.subscribe((queryChangeEvent, newQueryState) -> {
- final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<QueryChange>(0L,
+ final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<>(0L,
QueryChange.create(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
final Optional<StreamsQuery> expectedQueryState = Optional.of(
new StreamsQuery(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
@@ -210,7 +206,7 @@
//show listener not on the repo that query was added to is being notified as well.
final CountDownLatch repo2Latch = new CountDownLatch(1);
queries2.subscribe((queryChangeEvent, newQueryState) -> {
- final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<QueryChange>(0L,
+ final ChangeLogEntry<QueryChange> expected = new ChangeLogEntry<>(0L,
QueryChange.create(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
final Optional<StreamsQuery> expectedQueryState = Optional.of(
new StreamsQuery(queryChangeEvent.getEntry().getQueryId(), "query 2", true));
diff --git a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java
index cd78975..a5507a6 100644
--- a/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java
+++ b/extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java
@@ -23,6 +23,7 @@
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang.StringUtils;
import org.apache.rya.streams.api.entity.StreamsQuery;
import org.apache.rya.streams.api.exception.RyaStreamsException;
import org.apache.rya.streams.api.interactor.ListQueries;
@@ -112,12 +113,11 @@
sb.append("Queries in Rya Streams:\n");
sb.append("---------------------------------------------------------\n");
queries.forEach(query -> {
- sb.append("ID: ");
- sb.append(query.getQueryId());
- sb.append("\t\t");
- sb.append("Query: ");
- sb.append(query.getSparql());
- sb.append("\n");
+ sb.append("ID: ").append(query.getQueryId())
+ .append(" ")
+ .append("Is Active: ").append(query.isActive())
+ .append(StringUtils.rightPad("" + query.isActive(), 9))
+ .append("Query: ").append(query.getSparql()).append("\n");
});
return sb.toString();
}
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 ddaf647..7b311f6 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
@@ -136,7 +136,7 @@
final Set<String> topics = new HashSet<>();
topics.add( KafkaTopics.statementsTopic(params.ryaInstance) );
topics.add( KafkaTopics.queryResultsTopic(queryId) );
- KafkaTopics.createTopic(params.zookeeperServers, topics, 1, 1);
+ KafkaTopics.createTopics(params.zookeeperServers, topics, 1, 1);
// 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 095465c..989799a 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
@@ -55,7 +55,7 @@
}
/**
- * Get the Rya instance name from a Kafka topic name that has been used for a {@link QueryChnageLog}.
+ * Get the Rya instance name from a Kafka topic name that has been used for a {@link QueryChangeLog}.
* <p/>
* This is the inverse function of {@link #queryChangeLogTopic(String)}.
*
@@ -106,7 +106,7 @@
* @param partitions - The number of partitions that each of the topics will have.
* @param replicationFactor - The replication factor of the topics that are created.
*/
- public static void createTopic(
+ public static void createTopics(
final String zookeeperServers,
final Set<String> topicNames,
final int partitions,
diff --git a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
index 7ab7e90..8093951 100644
--- a/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java
@@ -83,7 +83,7 @@
try {
final TopologyBuilder topologyBuilder = topologyFactory.build(query.getSparql(), statementsTopic, resultsTopic, new RandomUUIDFactory());
return new KafkaStreams(topologyBuilder, new StreamsConfig(streamsProps));
- } catch (MalformedQueryException | TopologyBuilderException e) {
+ } catch (final MalformedQueryException | TopologyBuilderException e) {
throw new KafkaStreamsFactoryException("Could not create a KafkaStreams processing topology for query " + query.getQueryId(), e);
}
}
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
new file mode 100644
index 0000000..771e1c8
--- /dev/null
+++ b/extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/interactor/CreateKafkaTopic.java
@@ -0,0 +1,60 @@
+/**
+ * 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.streams.kafka.interactor;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Set;
+
+import org.apache.rya.streams.kafka.KafkaTopics;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Creates topics in Kafka.
+ */
+@DefaultAnnotation(NonNull.class)
+public class CreateKafkaTopic {
+
+ private final String zookeeperServers;
+
+ /**
+ * Constructs an instance of {@link CreateKafkaTopic}.
+ *
+ * @param zookeeperServers - The Zookeeper servers that are used to manage the Kafka instance. (not null)
+ */
+ public CreateKafkaTopic(final String zookeeperServers) {
+ this.zookeeperServers = requireNonNull(zookeeperServers);
+ }
+
+ /**
+ * Creates a set of Kafka topics for each topic that does not already exist.
+ *
+ * @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.
+ */
+ public void createTopics(
+ final Set<String> topicNames,
+ final int partitions,
+ final int replicationFactor) {
+ KafkaTopics.createTopics(zookeeperServers, topicNames, partitions, replicationFactor);
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/main/README.txt b/extras/rya.streams/query-manager/src/main/README.txt
index 3b2dbfe..93d5ac5 100644
--- a/extras/rya.streams/query-manager/src/main/README.txt
+++ b/extras/rya.streams/query-manager/src/main/README.txt
@@ -41,8 +41,11 @@
yum install -y ${project.artifactId}-${project.version}.noarch.rpm
3. Update the configuration file:
- Replace "[Kafka Broker Hostname]" with the IP address of the Kafka broker.
- Replace the Kafka port if using something other than the default of 9092.
+ A. Replace "[Zookeepers used to manage Kafka. E.g.: zoo1,zoo2,zoo3]" with
+ the zookeepers used to manage the Kafka cluster. It is a comma separated
+ list.
+ B. Replace "[Kafka Broker Hostname]" with the IP address of the Kafka broker.
+ C. Replace the Kafka port if using something other than the default of 9092.
4. Start the service:
systemctl start rya-streams-query-manager.service
diff --git a/extras/rya.streams/query-manager/src/main/config/configuration.xml b/extras/rya.streams/query-manager/src/main/config/configuration.xml
index 96da501..7077125 100644
--- a/extras/rya.streams/query-manager/src/main/config/configuration.xml
+++ b/extras/rya.streams/query-manager/src/main/config/configuration.xml
@@ -18,7 +18,7 @@
under the License.
-->
<queryManagerConfig>
- <!-- The Query Change Log Sources. The source defines a system where Rya
+ <!-- The Query Change Log Sources. The source defines a system where Rya
- Streams Query Change Logs are managed. The query manager will manage
- queries for all Rya instances whose change logs are stored within the
- source.
@@ -29,6 +29,14 @@
<port>9092</port>
</kafka>
</queryChangeLogSource>
+
+ <!-- The Query Executor. The executor defines a system for executing the
+ Rya Streams queries. -->
+ <queryExecutor>
+ <localKafkaStreams>
+ <zookeepers>[Zookeepers used to manage Kafka. E.g.: zoo1,zoo2,zoo3]</zookeepers>
+ </localKafkaStreams>
+ </queryExecutor>
<!-- This section defines performance related tuning values. Sensible
- default have been provided to simplify configuration.
diff --git a/extras/rya.streams/query-manager/src/main/config/log4j.xml b/extras/rya.streams/query-manager/src/main/config/log4j.xml
index 2021638..96ea56c 100644
--- a/extras/rya.streams/query-manager/src/main/config/log4j.xml
+++ b/extras/rya.streams/query-manager/src/main/config/log4j.xml
@@ -28,6 +28,23 @@
</layout>
</appender>
+ <!-- Kafka configuration configs are loud. -->
+ <logger name="org.apache.kafka.streams.StreamsConfig">
+ <level value="OFF"/>
+ </logger>
+ <logger name="org.apache.kafka.clients.consumer.ConsumerConfig">
+ <level value="OFF"/>
+ </logger>
+ <logger name="org.apache.kafka.clients.producer.ProducerConfig">
+ <level value="OFF"/>
+ </logger>
+
+ <!-- Change this level to DEBUG to see more information about what the
+ QueryManager is doing. -->
+ <logger name="org.apache.rya.streams.querymanager.QueryManager">
+ <level value="INFO"/>
+ </logger>
+
<root>
<level value="INFO" />
<appender-ref ref="console" />
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryChangeLogSource.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryChangeLogSource.java
index 73e5d12..eb5ca08 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryChangeLogSource.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryChangeLogSource.java
@@ -51,8 +51,9 @@
public void unsubscribe(final SourceListener listener);
/**
- * A listener that is notified when a {@link QueryChangeLog} has
- * been added or removed from a {@link QueryChangeLogSource}.
+ * A listener that is notified when a {@link QueryChangeLog} has been added or
+ * removed from a {@link QueryChangeLogSource}. The listener receives the only
+ * copy of the change log and is responsible for shutting it down.
*/
@DefaultAnnotation(NonNull.class)
public interface SourceListener {
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java
index 30b4538..e6bd800 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java
@@ -1,18 +1,20 @@
-/*
- * 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
+/**
+ * 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
+ * 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.
+ * 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.streams.querymanager;
@@ -20,16 +22,23 @@
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rya.streams.api.entity.StreamsQuery;
import org.apache.rya.streams.api.queries.ChangeLogEntry;
import org.apache.rya.streams.api.queries.InMemoryQueryRepository;
import org.apache.rya.streams.api.queries.QueryChange;
-import org.apache.rya.streams.api.queries.QueryChange.ChangeType;
import org.apache.rya.streams.api.queries.QueryChangeLog;
import org.apache.rya.streams.api.queries.QueryChangeLogListener;
import org.apache.rya.streams.api.queries.QueryRepository;
@@ -38,8 +47,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.util.concurrent.AbstractIdleService;
+import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.AbstractScheduledService.Scheduler;
+import com.google.common.util.concurrent.AbstractService;
+import com.google.common.util.concurrent.UncheckedExecutionException;
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
import edu.umd.cs.findbugs.annotations.NonNull;
@@ -51,205 +62,823 @@
* instances/rya streams instances.
*/
@DefaultAnnotation(NonNull.class)
-public class QueryManager extends AbstractIdleService {
- private static final Logger LOG = LoggerFactory.getLogger(QueryManager.class);
-
- private final QueryExecutor queryExecutor;
- private final Scheduler scheduler;
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
/**
- * Map of Rya Instance name to {@link QueryRepository}.
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
*/
- private final Map<String, QueryRepository> queryRepos = new HashMap<>();
+ private final QueryChangeLogSource changeLogSource;
- private final ReentrantLock lock = new ReentrantLock();
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
- private final QueryChangeLogSource source;
+ /**
+ * How long blocking operations will be attempted before potentially trying again.
+ */
+ private final long blockingValue;
+
+ /**
+ * The units for {@link #blockingValue}.
+ */
+ private final TimeUnit blockingUnits;
+
+ /**
+ * Used to inform threads that the application is shutting down, so they must stop work.
+ */
+ private final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ /**
+ * This thread pool manages the two thread used to work the {@link LogEvent}s
+ * and the {@link QueryEvent}s.
+ */
+ private final ExecutorService executor = Executors.newFixedThreadPool(2);
/**
* Creates a new {@link QueryManager}.
*
* @param queryExecutor - Runs the active {@link StreamsQuery}s. (not null)
* @param source - The {@link QueryChangeLogSource} of QueryChangeLogs. (not null)
- * @param scheduler - The {@link Scheduler} used to discover query changes
- * within the {@link QueryChangeLog}s (not null)
+ * @param blockingValue - How long blocking operations will try before looping. (> 0)
+ * @param blockingUnits - The units of the {@code blockingValue}. (not null)
*/
- public QueryManager(final QueryExecutor queryExecutor, final QueryChangeLogSource source, final Scheduler scheduler) {
- this.source = requireNonNull(source);
+ public QueryManager(
+ final QueryExecutor queryExecutor,
+ final QueryChangeLogSource source,
+ final long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
this.queryExecutor = requireNonNull(queryExecutor);
- this.scheduler = requireNonNull(scheduler);
- }
-
- /**
- * Starts running a query.
- *
- * @param ryaInstanceName - The Rya instance the query belongs to. (not null)
- * @param query - The query to run.(not null)
- */
- private void runQuery(final String ryaInstanceName, final StreamsQuery query) {
- requireNonNull(ryaInstanceName);
- requireNonNull(query);
- LOG.info("Starting Query: " + query.toString() + " on the rya instance: " + ryaInstanceName);
-
- try {
- queryExecutor.startQuery(ryaInstanceName, query);
- } catch (final QueryExecutorException e) {
- LOG.error("Failed to start query.", e);
- }
- }
-
- /**
- * Stops the specified query from running.
- *
- * @param queryId - The ID of the query to stop running. (not null)
- */
- private void stopQuery(final UUID queryId) {
- requireNonNull(queryId);
-
- LOG.info("Stopping query: " + queryId.toString());
-
- try {
- queryExecutor.stopQuery(queryId);
- } catch (final QueryExecutorException e) {
- LOG.error("Failed to stop query.", e);
- }
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
}
@Override
- protected void startUp() throws Exception {
- lock.lock();
+ protected void doStart() {
+ log.info("Starting a QueryManager.");
+
+ // A work queue of discovered Query Change Logs that need to be handled.
+ // This queue exists so that the source notifying thread may be released
+ // immediately instead of calling into blocking functions.
+ final BlockingQueue<LogEvent> logEvents = new ArrayBlockingQueue<>(1024);
+
+ // A work queue of discovered Query Changes from the monitored Query Change Logs
+ // that need to be handled. This queue exists so that the Query Repository notifying
+ // thread may be released immediately instead of calling into blocking functions.
+ final BlockingQueue<QueryEvent> queryEvents = new ArrayBlockingQueue<>(1024);
+
try {
- LOG.info("Starting Query Manager.");
+ // Start up a LogEventWorker using the executor service.
+ executor.submit(new LogEventWorker(logEvents, queryEvents, blockingValue, blockingUnits, shutdownSignal));
+
+ // Start up a QueryEvent Worker using the executor service.
+ executor.submit(new QueryEventWorker(queryEvents, queryExecutor, blockingValue, blockingUnits, shutdownSignal));
+
+ // Start up the query execution framework.
queryExecutor.startAndWait();
- source.startAndWait();
- // subscribe to the sources to be notified of changes.
- source.subscribe(new QueryManagerSourceListener());
- } finally {
- lock.unlock();
+ // Startup the source that discovers new Query Change Logs.
+ changeLogSource.startAndWait();
+
+ // Subscribe the source a listener that writes to the LogEventWorker's work queue.
+ changeLogSource.subscribe(new LogEventWorkGenerator(logEvents, blockingValue, blockingUnits, shutdownSignal));
+ } catch(final RejectedExecutionException | UncheckedExecutionException e) {
+ log.error("Could not start up a QueryManager.", e);
+ notifyFailed(e);
}
+
+ // Notify the service was successfully started.
+ notifyStarted();
+
+ log.info("QueryManager has finished starting.");
}
@Override
- protected void shutDown() throws Exception {
- lock.lock();
+ protected void doStop() {
+ log.info("Stopping a QueryManager.");
+
+ // Set the shutdown flag so that all components that rely on that signal will stop processing.
+ shutdownSignal.set(true);
+
+ // Stop the workers and wait for them to die.
+ executor.shutdownNow();
try {
- LOG.info("Stopping Query Manager.");
- source.stopAndWait();
- queryExecutor.stopAndWait();
- } finally {
- lock.unlock();
+ if(!executor.awaitTermination(10, TimeUnit.SECONDS)) {
+ log.warn("Waited 10 seconds for the worker threads to die, but they are still running.");
+ }
+ } catch (final InterruptedException e) {
+ log.warn("Waited 10 seconds for the worker threads to die, but they are still running.");
}
+
+ // Stop the source of new Change Logs.
+ try {
+ changeLogSource.stopAndWait();
+ } catch(final UncheckedExecutionException e) {
+ log.warn("Could not stop the Change Log Source.", e);
+ }
+
+ // Stop the query execution framework.
+ try {
+ queryExecutor.stopAndWait();
+ } catch(final UncheckedExecutionException e) {
+ log.warn("Could not stop the Query Executor", e);
+ }
+
+ // Notify the service was successfully stopped.
+ notifyStopped();
+
+ log.info("QueryManager has finished stopping.");
}
/**
- * An implementation of {@link QueryChangeLogListener} for the
- * {@link QueryManager}.
- * <p>
- * When notified of a {@link ChangeType} performs one of the following:
- * <li>{@link ChangeType#CREATE}: Creates a new query using the
- * {@link QueryExecutor} provided to the {@link QueryManager}</li>
- * <li>{@link ChangeType#DELETE}: Deletes a running query by stopping the
- * {@link QueryExecutor} service of the queryID in the event</li>
- * <li>{@link ChangeType#UPDATE}: If the query is running and the update is
- * to stop the query, stops the query. Otherwise, if the query is not
- * running, it is removed.</li>
+ * Offer a unit of work to a blocking queue until it is either accepted, or the
+ * shutdown signal is set.
+ *
+ * @param workQueue - The blocking work queue to write to. (not null)
+ * @param event - The event that will be offered to the work queue. (not null)
+ * @param offerValue - How long to wait when offering new work.
+ * @param offerUnits - The unit for the {@code offerValue}. (not null)
+ * @param shutdownSignal - Used to signal application shutdown has started, so
+ * this method may terminate without ever placing the event on the queue. (not null)
+ * @return {@code true} if the evet nwas added to the queue, otherwise false.
*/
- private class QueryExecutionForwardingListener implements QueryChangeLogListener {
- private final String ryaInstanceName;
+ private static <T> boolean offerUntilAcceptedOrShutdown(
+ final BlockingQueue<T> workQueue,
+ final T event,
+ final long offerValue,
+ final TimeUnit offerUnits,
+ final AtomicBoolean shutdownSignal) {
+ requireNonNull(workQueue);
+ requireNonNull(event);
+ requireNonNull(shutdownSignal);
+
+ boolean submitted = false;
+ while(!submitted && !shutdownSignal.get()) {
+ try {
+ submitted = workQueue.offer(event, offerValue, offerUnits);
+ if(!submitted) {
+ log.debug("An event could not be added to a work queue after waiting 5 seconds. Trying again...");
+ }
+ } catch (final InterruptedException e) {
+ log.debug("An event could not be added to a work queue after waiting 5 seconds. Trying again...");
+ }
+ }
+ return submitted;
+ }
+
+ /**
+ * An observation that a {@link QueryChangeLog} was created within or
+ * removed from a {@link QueryChangeLogSource}.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEvent {
/**
- * Creates a new {@link QueryExecutionForwardingListener}.
- *
- * @param ryaInstanceName - The rya instance the query change is
- * performed on. (not null)
+ * The types of events that may be observed.
*/
- public QueryExecutionForwardingListener(final String ryaInstanceName) {
- this.ryaInstanceName = requireNonNull(ryaInstanceName);
+ static enum LogEventType {
+ /**
+ * A {@link QueryChangeLog} was created within a {@link QueryChangeLogSource}.
+ */
+ CREATE,
+
+ /**
+ * A {@link QueryChangeLog} was deleted from a {@link QueryChangeLogSource}.
+ */
+ DELETE;
+ }
+
+ private final String ryaInstance;
+ private final LogEventType eventType;
+ private final Optional<QueryChangeLog> log;
+
+ /**
+ * Constructs an instance of {@link LogEvent}.
+ *
+ * @param ryaInstance - The Rya Instance the log is/was for. (not null)
+ * @param eventType - The type of event that was observed. (not null)
+ * @param log - The log if this is a create event. (not null)
+ */
+ private LogEvent(final String ryaInstance, final LogEventType eventType, final Optional<QueryChangeLog> log) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.eventType = requireNonNull(eventType);
+ this.log = requireNonNull(log);
+ }
+
+ /**
+ * @return The Rya Instance whose log was either created or deleted.
+ */
+ public String getRyaInstanceName() {
+ return ryaInstance;
+ }
+
+ /**
+ * @return The type of event that was observed.
+ */
+ public LogEventType getEventType() {
+ return eventType;
+ }
+
+ /**
+ * @return The {@link QueryChangeLog} if this is a CREATE event.
+ */
+ public Optional<QueryChangeLog> getQueryChangeLog() {
+ return log;
}
@Override
- public void notify(final ChangeLogEntry<QueryChange> queryChangeEvent, final Optional<StreamsQuery> newQueryState) {
- LOG.debug("New query change event.");
- final QueryChange entry = queryChangeEvent.getEntry();
+ public String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
- lock.lock();
- try {
+ /**
+ * Make a {@link LogEvent} that indicates a {@link QueryChangeLog} was created within a
+ * {@link QueryChangeLogSource}.
+ *
+ * @param ryaInstance - The Rya Instance the created log is for. (not null)
+ * @param log - The created {@link QueryChangeLog. (not null)
+ * @return A {@link LogEvent} built using the provided values.
+ */
+ public static LogEvent create(final String ryaInstance, final QueryChangeLog log) {
+ return new LogEvent(ryaInstance, LogEventType.CREATE, Optional.of(log));
+ }
- switch (entry.getChangeType()) {
- case CREATE:
- if(!newQueryState.isPresent()) {
- LOG.error("The query with ID: " + entry.getQueryId() + " must be present with the change to be created.");
- LOG.debug("newQueryState is not allowed to be absent with a CREATE QueryChange, there might be a bug in the QueryRepository.");
- } else {
- runQuery(ryaInstanceName, newQueryState.get());
- }
- break;
- case DELETE:
- stopQuery(entry.getQueryId());
- break;
- case UPDATE:
- if (!newQueryState.isPresent()) {
- LOG.error("The query with ID: " + entry.getQueryId() + " must be provided with the update, cannot perform update.");
- LOG.debug("newQueryState is not allowed to be absent with a UPDATE QueryChange, there might be a bug in the QueryRepository.");
- } else {
- final StreamsQuery updatedQuery = newQueryState.get();
- if (updatedQuery.isActive()) {
- runQuery(ryaInstanceName, updatedQuery);
- LOG.info("Starting query: " + updatedQuery.toString());
- } else {
- stopQuery(updatedQuery.getQueryId());
- LOG.info("Stopping query: " + updatedQuery.toString());
- }
- }
- break;
- }
- } finally {
- lock.unlock();
- }
+ /**
+ * Make a {@link LogEvent} that indicates a {@link QueryChangeLog} was deleted from
+ * a {@link QueryChangeLogSource}.
+ *
+ * @param ryaInstance - The Rya Instance whose log was deleted. (not null)
+ * @return A {@link LogEvent} built using the provided values.
+ */
+ public static LogEvent delete(final String ryaInstance) {
+ return new LogEvent(ryaInstance, LogEventType.DELETE, Optional.empty());
}
}
/**
- * Listener used by the {@link QueryManager} to be notified when
- * {@link QueryChangeLog}s are created or deleted.
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
*/
- private class QueryManagerSourceListener implements SourceListener {
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEvent {
+
+ /**
+ * The type of events that may be observed.
+ */
+ public static enum QueryEventType {
+ /**
+ * Indicates a {@link StreamsQuery} needs to be executing.
+ */
+ EXECUTING,
+
+ /**
+ * Indicates a {@link StreamsQuery} needs to be stopped.
+ */
+ STOPPED,
+
+ /**
+ * Indicates all {@link StreamsQuery}s for a Rya instance need to be stopped.
+ */
+ STOP_ALL;
+ }
+
+ private final String ryaInstance;
+ private final QueryEventType type;
+ private final Optional<UUID> queryId;
+ private final Optional<StreamsQuery> query;
+
+ /**
+ * Constructs an instance of {@link QueryEvent}.
+ *
+ * @param ryaInstance - The Rya instance that generated the event. (not null)
+ * @param type - Indicates whether the query needs to be executing or not. (not null)
+ * @param queryId - If stopped, the ID of the query that must not be running. (not null)
+ * @param query - If executing, the StreamsQuery that defines what should be executing. (not null)
+ */
+ private QueryEvent(
+ final String ryaInstance,
+ final QueryEventType type,
+ final Optional<UUID> queryId,
+ final Optional<StreamsQuery> query) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.type = requireNonNull(type);
+ this.queryId = requireNonNull(queryId);
+ this.query = requireNonNull(query);
+ }
+
+ /**
+ * @return The Rya instance that generated the event.
+ */
+ public String getRyaInstance() {
+ return ryaInstance;
+ }
+
+ /**
+ * @return Indicates whether the query needs to be executing or not.
+ */
+ public QueryEventType getType() {
+ return type;
+ }
+
+ /**
+ * @return If stopped, the ID of the query that must not be running. Otherwise absent.
+ */
+ public Optional<UUID> getQueryId() {
+ return queryId;
+ }
+
+ /**
+ * @return If executing, the StreamsQuery that defines what should be executing. Otherwise absent.
+ */
+ public Optional<StreamsQuery> getStreamsQuery() {
+ return query;
+ }
+
@Override
- public void notifyCreate(final String ryaInstanceName, final QueryChangeLog log) {
- lock.lock();
- try {
- LOG.info("Discovered new Query Change Log for Rya Instance " + ryaInstanceName + ".");
- final QueryRepository repo = new InMemoryQueryRepository(log, scheduler);
- repo.startAndWait();
- final Set<StreamsQuery> queries = repo.subscribe(new QueryExecutionForwardingListener(ryaInstanceName));
- queries.forEach(query -> {
- if (query.isActive()) {
- try {
- queryExecutor.startQuery(ryaInstanceName, query);
- } catch (IllegalStateException | QueryExecutorException e) {
- LOG.error("Unable to start query for rya instance " + ryaInstanceName, e);
- }
- }
- });
- queryRepos.put(ryaInstanceName, repo);
- } finally {
- lock.unlock();
+ public int hashCode() {
+ return Objects.hash(ryaInstance, type, queryId, query);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if(o instanceof QueryEvent) {
+ final QueryEvent other = (QueryEvent) o;
+ return Objects.equals(ryaInstance, other.ryaInstance) &&
+ Objects.equals(type, other.type) &&
+ Objects.equals(queryId, other.queryId) &&
+ Objects.equals(query, other.query);
}
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder string = new StringBuilder();
+ string.append("Query Event {\n")
+ .append(" Rya Instance: ").append(ryaInstance).append(",\n")
+ .append(" Type: ").append(type).append(",\n");
+ switch(type) {
+ case EXECUTING:
+ append(string, query.get());
+ break;
+ case STOPPED:
+ string.append(" Query ID: ").append(queryId.get()).append("\n");
+ break;
+ case STOP_ALL:
+ break;
+ default:
+ // Default to showing everything that is in the object.
+ string.append(" Query ID: ").append(queryId.get()).append("\n");
+ append(string, query.get());
+ break;
+ }
+ string.append("}");
+ return string.toString();
+ }
+
+ private void append(final StringBuilder string, final StreamsQuery query) {
+ requireNonNull(string);
+ requireNonNull(query);
+ string.append(" Streams Query {\n")
+ .append(" Query ID: ").append(query.getQueryId()).append(",\n")
+ .append(" Is Active: ").append(query.isActive()).append(",\n")
+ .append(" SPARQL: ").append(query.getSparql()).append("\n")
+ .append(" }");
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates a query needs to be executing.
+ *
+ * @param ryaInstance - The Rya instance that generated the event. (not null)
+ * @param query - The StreamsQuery that defines what should be executing. (not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent executing(final String ryaInstance, final StreamsQuery query) {
+ return new QueryEvent(ryaInstance, QueryEventType.EXECUTING, Optional.empty(), Optional.of(query));
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates a query needs to be stopped.
+ *
+ * @param ryaInstance - The Rya instance that generated the event. (not null)
+ * @param queryId - The ID of the query that must not be running. (not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent stopped(final String ryaInstance, final UUID queryId) {
+ return new QueryEvent(ryaInstance, QueryEventType.STOPPED, Optional.of(queryId), Optional.empty());
+ }
+
+ /**
+ * Create a {@link QueryEvent} that indicates all queries for a Rya instance needs to be stopped.
+ *
+ * @param ryaInstance - The Rya instance that generated the event. (not null)
+ * @return A {@link QueryEvent} built using the provided values.
+ */
+ public static QueryEvent stopALL(final String ryaInstance) {
+ return new QueryEvent(ryaInstance, QueryEventType.STOP_ALL, Optional.empty(), Optional.empty());
+ }
+ }
+
+ /**
+ * Listens to a {@link QueryChangeLogSource} and adds observations to the provided
+ * work queue. It does so until the provided shutdown signal is set.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEventWorkGenerator implements SourceListener {
+
+ private final BlockingQueue<LogEvent> workQueue;
+ private final AtomicBoolean shutdownSignal;
+ private final long offerValue;
+ private final TimeUnit offerUnits;
+
+ /**
+ * Constructs an instance of {@link QueryManagerSourceListener}.
+ *
+ * @param workQueue - A blocking queue that will have {@link LogEvent}s offered to it. (not null)
+ * @param offerValue - How long to wait when offering new work.
+ * @param offerUnits - The unit for the {@code offerValue}. (not null)
+ * @param shutdownSignal - Indicates to this listener that it needs to stop adding events
+ * to the work queue because the application is shutting down. (not null)
+ */
+ public LogEventWorkGenerator(
+ final BlockingQueue<LogEvent> workQueue,
+ final long offerValue,
+ final TimeUnit offerUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.workQueue = requireNonNull(workQueue);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ this.offerValue = offerValue;
+ this.offerUnits = requireNonNull(offerUnits);
+ }
+
+ @Override
+ public void notifyCreate(final String ryaInstanceName, final QueryChangeLog changeLog) {
+ log.info("A new Query Change Log has been discovered for Rya Instance " + ryaInstanceName + ". All " +
+ "queries that are set to active within it will be started.");
+
+ // Create an event that summarizes this notification.
+ final LogEvent event = LogEvent.create(ryaInstanceName, changeLog);
+
+ // Offer it to the worker until there is room for it in the work queue, or we are shutting down.
+ offerUntilAcceptedOrShutdown(workQueue, event, offerValue, offerUnits, shutdownSignal);
}
@Override
public void notifyDelete(final String ryaInstanceName) {
- lock.lock();
+ log.info("The Query Change Log for Rya Instance " + ryaInstanceName + " has been deleted. All of the " +
+ "queries related to that instance will be stopped.");
+
+ // Create an event that summarizes this notification.
+ final LogEvent event = LogEvent.delete(ryaInstanceName);
+
+ // Offer it to the worker until there is room for it in the work queue, or we are shutting down.
+ offerUntilAcceptedOrShutdown(workQueue, event, offerValue, offerUnits, shutdownSignal);
+ }
+ }
+
+ /**
+ * Processes a work queue of {@link LogEvent}s.
+ * <p/>
+ * Whenever a new log has been created, then it registers a {@link QueryEventWorkGenerator}
+ * that generates {@link QueryEvent}s based on the content and updates to the discovered
+ * {@link QueryChagneLog}.
+ * <p/>
+ * Whenever a log is deleted, then the generator is stopped and a stop all {@link QueryEvent}
+ * is written to the work queue.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class LogEventWorker implements Runnable {
+
+ /**
+ * A map of Rya Instance name to he Query Repository for that instance.
+ */
+ private final Map<String, QueryRepository> repos = new HashMap<>();
+
+ private final BlockingQueue<LogEvent> logWorkQueue;
+ private final BlockingQueue<QueryEvent> queryWorkQueue;
+ private final long blockingValue;
+ private final TimeUnit blockingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link LogEventWorker}.
+ *
+ * @param logWorkQueue - A queue of {@link LogEvent}s that will be worked by this object. (not null)
+ * @param queryWorkQueue - A queue where {@link QueryEvent}s will be placed by this object. (not null)
+ * @param blockingValue - How long to wait when polling/offering new work.
+ * @param blockingUnits - The unit for the {@code blockingValue}. (not null)
+ * @param shutdownSignal - Indicates when the application has been shutdown, so the executing thread
+ * may exit the {@link #run()} method. (not null)
+ */
+ public LogEventWorker(
+ final BlockingQueue<LogEvent> logWorkQueue,
+ final BlockingQueue<QueryEvent> queryWorkQueue,
+ final long blockingValue,
+ final TimeUnit blockingUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.logWorkQueue = requireNonNull(logWorkQueue);
+ this.queryWorkQueue = requireNonNull(queryWorkQueue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ }
+
+ @Override
+ public void run() {
+ // Run until the shutdown signal is set.
+ while(!shutdownSignal.get()) {
+ try {
+ // Pull a unit of work from the queue.
+ log.debug("LogEventWorker - Polling the work queue for a new LogEvent.");
+ final LogEvent logEvent = logWorkQueue.poll(blockingValue, blockingUnits);
+ if(logEvent == null) {
+ // Poll again if nothing was found.
+ continue;
+ }
+
+ log.info("LogEventWorker - handling: \n" + logEvent);
+ final String ryaInstance = logEvent.getRyaInstanceName();
+
+ switch(logEvent.getEventType()) {
+ case CREATE:
+ // If we see a create message for a Rya Instance we are already maintaining,
+ // then don't do anything.
+ if(repos.containsKey(ryaInstance)) {
+ log.warn("LogEventWorker - A repository is already being managed for the Rya Instance " +
+ ryaInstance + ". This message will be ignored.");
+ continue;
+ }
+
+ // Create and start a QueryRepository for the discovered log. Hold onto the repository
+ // so that it may be shutdown later.
+ final Scheduler scheduler = Scheduler.newFixedRateSchedule(0, blockingValue, blockingUnits);
+ final QueryRepository repo = new InMemoryQueryRepository(logEvent.getQueryChangeLog().get(), scheduler);
+ repo.startAndWait();
+ repos.put(ryaInstance, repo);
+
+ // Subscribe a worker that adds the Query Events to the queryWorkQueue queue.
+ // A count down latch is used to ensure the returned set of queries are handled
+ // prior to any notifications from the repository.
+ final CountDownLatch subscriptionWorkFinished = new CountDownLatch(1);
+ final QueryEventWorkGenerator queryWorkGenerator =
+ new QueryEventWorkGenerator(ryaInstance, subscriptionWorkFinished, queryWorkQueue,
+ blockingValue, blockingUnits, shutdownSignal);
+
+ log.debug("LogEventWorker - Setting up a QueryWorkGenerator...");
+ final Set<StreamsQuery> queries = repo.subscribe(queryWorkGenerator);
+ log.debug("LogEventWorker - Finished setting up a QueryWorkGenerator.");
+
+ // Handle the view of the queries within the repository as it existed when
+ // the subscription was registered.
+ queries.stream()
+ .forEach(query -> {
+ // Create a QueryEvent that represents the active state of the existing query.
+ final QueryEvent queryEvent = query.isActive() ?
+ QueryEvent.executing(ryaInstance, query) : QueryEvent.stopped(ryaInstance, query.getQueryId());
+ log.debug("LogEventWorker - offering: " + queryEvent);
+
+ // Offer it to the worker until there is room for it in the work queue, or we are shutting down.
+ offerUntilAcceptedOrShutdown(queryWorkQueue, queryEvent, blockingValue, blockingUnits, shutdownSignal);
+ });
+
+ // Indicate the subscription work is finished so that the registered listener may start
+ // adding work to the queue.
+ log.info("LogEventWorker - Counting down the subscription work latch.");
+ subscriptionWorkFinished.countDown();
+ break;
+
+ case DELETE:
+ if(repos.containsKey(ryaInstance)) {
+ // Shut down the query repository for the Rya instance. This ensures the listener will
+ // not receive any more work that needs to be done.
+ final QueryRepository deletedRepo = repos.remove(ryaInstance);
+ deletedRepo.stopAndWait();
+
+ // Add work that stops all of the queries related to the instance.
+ final QueryEvent stopAllEvent = QueryEvent.stopALL(ryaInstance);
+ offerUntilAcceptedOrShutdown(queryWorkQueue, stopAllEvent, blockingValue, blockingUnits, shutdownSignal);
+ }
+ break;
+ }
+ } catch (final InterruptedException e) {
+ log.debug("LogEventWorker did not see any new events over the past 5 seconds. Polling again...");
+ }
+ }
+
+ log.info("LogEventWorker shutting down...");
+
+ // Shutdown all of the QueryRepositories that were started.
+ repos.values().forEach(repo -> repo.stopAndWait());
+
+ log.info("LogEventWorker shut down.");
+ }
+ }
+
+ /**
+ * Listens to a {@link QueryRepository} and adds observations to the provided work queue.
+ * It does so until the provided shutdown signal is set.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEventWorkGenerator implements QueryChangeLogListener {
+
+ private final String ryaInstance;
+ private final CountDownLatch subscriptionWorkFinished;
+ private final BlockingQueue<QueryEvent> queryWorkQueue;
+ private final long blockingValue;
+ private final TimeUnit blockingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link QueryEventWorkGenerator}.
+ *
+ * @param ryaInstance - The rya instance whose log this objects is watching. (not null)
+ * @param subscriptionWorkFinished - Indicates when work that needs to be completed before this
+ * listener handles notifications is completed. (not null)
+ * @param queryWorkQueue - A queue where {@link QueryEvent}s will be placed by this object. (not null)
+ * @param blockingValue - How long to wait when polling/offering new work.
+ * @param blockingUnits - The unit for the {@code blockingValue}. (not null)
+ * @param shutdownSignal - Indicates to this listener that it needs to stop adding events
+ * to the work queue because the application is shutting down. (not null)
+ */
+ public QueryEventWorkGenerator(
+ final String ryaInstance,
+ final CountDownLatch subscriptionWorkFinished,
+ final BlockingQueue<QueryEvent> queryWorkQueue,
+ final long blockingValue,
+ final TimeUnit blockingUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.ryaInstance = requireNonNull(ryaInstance);
+ this.subscriptionWorkFinished = requireNonNull(subscriptionWorkFinished);
+ this.queryWorkQueue = requireNonNull(queryWorkQueue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ }
+
+ @Override
+ public void notify(final ChangeLogEntry<QueryChange> queryChangeEvent, final Optional<StreamsQuery> newQueryState) {
+ requireNonNull(queryChangeEvent);
+ requireNonNull(newQueryState);
+
+ // Wait for the subscription work to be finished.
try {
- LOG.info("Notified of deleting QueryChangeLog, stopping all queries belonging to the change log for "
- + ryaInstanceName + ".");
- queryExecutor.stopAll(ryaInstanceName);
- } catch (final QueryExecutorException e) {
- LOG.error("Failed to stop all queries belonging to: " + ryaInstanceName, e);
- } finally {
- lock.unlock();
+ log.debug("Waiting for Subscription Work Finished latch to release...");
+ while(!shutdownSignal.get() && !subscriptionWorkFinished.await(blockingValue, blockingUnits)) {
+ log.debug("Still waiting...");
+ }
+ log.debug("Subscription Work Finished latch to released.");
+ } catch (final InterruptedException e) {
+ log.warn("Interrupted while waiting for the Subscription Work Finished latch to be " +
+ "released. Shutting down?", e);
+ }
+
+ // If we left the loop because of a shutdown, return immediately.
+ if(shutdownSignal.get()) {
+ log.debug("Not processing notification. Shutting down.");
+ return;
+ }
+
+ // Generate work from the notification.
+ final QueryChange change = queryChangeEvent.getEntry();
+ switch(change.getChangeType()) {
+ case CREATE:
+ if(newQueryState.isPresent()) {
+ log.info("Rya Instance " + ryaInstance + " created Rya Streams query " + newQueryState + ".");
+ final StreamsQuery newQuery = newQueryState.get();
+ if(newQuery.isActive()) {
+ final QueryEvent executeNewQuery = QueryEvent.executing(ryaInstance, newQuery);
+ offerUntilAcceptedOrShutdown(queryWorkQueue, executeNewQuery, blockingValue, blockingUnits, shutdownSignal);
+ }
+ } else {
+ log.error("Received a CREATE QueryChange for Rya Instance: " + ryaInstance +
+ ", Query ID: " + change.getQueryId() + ", but the QueryRepository did not supply a " +
+ "StreamsQuery representing the created query. The query will not be processed.");
+ }
+ break;
+
+ case DELETE:
+ final UUID deletedQueryId = change.getQueryId();
+ log.info("Rya Instance " + ryaInstance + " deleted Rya Streams query with ID " + deletedQueryId);
+ final QueryEvent stopDeletedQuery = QueryEvent.stopped(ryaInstance, deletedQueryId);
+ offerUntilAcceptedOrShutdown(queryWorkQueue, stopDeletedQuery, blockingValue, blockingUnits, shutdownSignal);
+ break;
+
+ case UPDATE:
+ if(newQueryState.isPresent()) {
+ final StreamsQuery updatedQuery = newQueryState.get();
+ if(updatedQuery.isActive()) {
+ log.info("Rya Instance " + ryaInstance + " updated Rya Streams query with ID " +
+ updatedQuery.getQueryId() + " to be active.");
+ final QueryEvent executeUpdatedQuery = QueryEvent.executing(ryaInstance, updatedQuery);
+ offerUntilAcceptedOrShutdown(queryWorkQueue, executeUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);
+ } else {
+ log.info("Rya Instance " + ryaInstance + " updated Rya Streams query with ID " +
+ updatedQuery.getQueryId() + " to be inactive.");
+ final QueryEvent stopUpdatedQuery = QueryEvent.stopped(ryaInstance, updatedQuery.getQueryId());
+ offerUntilAcceptedOrShutdown(queryWorkQueue, stopUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);
+ }
+ } else {
+ log.error("Received an UPDATE QueryChange for Rya Instance: " + ryaInstance +
+ ", Query ID: " + change.getQueryId() + ", but the QueryRepository did not supply a " +
+ "StreamsQuery representing the created query. The query will not be processed.");
+ }
+ break;
}
}
}
-}
+
+ /**
+ * Processes a work queue of {@link QueryEvent}s.
+ * <p/>
+ * Each type of event maps the to corresponding method on {@link QueryExecutor} that is called into.
+ */
+ @DefaultAnnotation(NonNull.class)
+ static class QueryEventWorker implements Runnable {
+
+ private final BlockingQueue<QueryEvent> workQueue;
+ private final QueryExecutor queryExecutor;
+ private final long pollingValue;
+ private final TimeUnit pollingUnits;
+ private final AtomicBoolean shutdownSignal;
+
+ /**
+ * Constructs an instance of {@link QueryEventWorker}.
+ *
+ * @param workQueue - A queue of {@link QueryEvent}s that will be worked by this object. (not null)
+ * @param queryExecutor - Responsible for executing the {@link StreamsQuery}s. (not null)
+ * @param pollingValue - How long to wait when polling for new work.
+ * @param pollingUnits - The units for the {@code pollingValue}. (not null)
+ * @param shutdownSignal - Indicates when the application has been shutdown, so the executing thread
+ * may exit the {@link #run()} method. (not null)
+ */
+ public QueryEventWorker(
+ final BlockingQueue<QueryEvent> workQueue,
+ final QueryExecutor queryExecutor,
+ final long pollingValue,
+ final TimeUnit pollingUnits,
+ final AtomicBoolean shutdownSignal) {
+ this.workQueue = requireNonNull(workQueue);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ this.pollingValue = pollingValue;
+ this.pollingUnits = requireNonNull(pollingUnits);
+ this.shutdownSignal = requireNonNull(shutdownSignal);
+ }
+
+ @Override
+ public void run() {
+ log.info("QueryEventWorker starting.");
+
+ // Run until the shutdown signal is set.
+ while(!shutdownSignal.get()) {
+ // Pull a unit of work from the queue.
+ try {
+ log.debug("Polling the work queue for a new QueryEvent.");
+ final QueryEvent event = workQueue.poll(pollingValue, pollingUnits);
+ if(event == null) {
+ // Poll again if nothing was found.
+ continue;
+ }
+
+ log.info("QueryEventWorker handling:\n" + event);
+
+ // Ensure the state within the executor matches the query event's state.
+ switch(event.getType()) {
+ case EXECUTING:
+ try {
+ queryExecutor.startQuery(event.getRyaInstance(), event.getStreamsQuery().get());
+ } catch (final IllegalStateException | QueryExecutorException e) {
+ log.error("Could not start a query represented by the following work: " + event, e);
+ }
+ break;
+
+ case STOPPED:
+ try {
+ queryExecutor.stopQuery(event.getQueryId().get());
+ } catch (final IllegalStateException | QueryExecutorException e) {
+ log.error("Could not stop a query represented by the following work: " + event, e);
+ }
+ break;
+
+ case STOP_ALL:
+ try {
+ queryExecutor.stopAll(event.getRyaInstance());
+ } catch (final IllegalStateException | QueryExecutorException e) {
+ log.error("Could not stop all queries represented by the following work: " + event, e);
+ }
+ break;
+ }
+ } catch (final InterruptedException e) {
+ log.debug("QueryEventWorker interrupted. Probably shutting down.");
+ }
+ }
+ log.info("QueryEventWorker shut down.");
+ }
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.java
index 515d699..04a0382 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.java
@@ -33,6 +33,7 @@
import org.apache.commons.daemon.DaemonInitException;
import org.apache.rya.streams.kafka.KafkaStreamsFactory;
import org.apache.rya.streams.kafka.SingleThreadKafkaStreamsFactory;
+import org.apache.rya.streams.kafka.interactor.CreateKafkaTopic;
import org.apache.rya.streams.querymanager.kafka.KafkaQueryChangeLogSource;
import org.apache.rya.streams.querymanager.kafka.LocalQueryExecutor;
import org.apache.rya.streams.querymanager.xml.Kafka;
@@ -91,7 +92,7 @@
// Unmarshall the configuration file into an object.
final QueryManagerConfig config;
- try(InputStream stream = Files.newInputStream(configFile)) {
+ try(final InputStream stream = Files.newInputStream(configFile)) {
config = QueryManagerConfigUnmarshaller.unmarshall(stream);
} catch(final JAXBException | SAXException e) {
throw new DaemonInitException("Unable to marshall the configuration XML file: " + configFile, e);
@@ -110,11 +111,12 @@
final QueryChangeLogSource source = new KafkaQueryChangeLogSource(kafka.getHostname(), kafka.getPort(), scheduler);
// Initialize a QueryExecutor.
+ final String zookeeperServers = config.getQueryExecutor().getLocalKafkaStreams().getZookeepers();
final KafkaStreamsFactory streamsFactory = new SingleThreadKafkaStreamsFactory(kafka.getHostname() + ":" + kafka.getPort());
- final QueryExecutor queryExecutor = new LocalQueryExecutor(streamsFactory);
+ final QueryExecutor queryExecutor = new LocalQueryExecutor(new CreateKafkaTopic(zookeeperServers), streamsFactory);
// Initialize the QueryManager using the configured resources.
- manager = new QueryManager(queryExecutor, source, scheduler);
+ manager = new QueryManager(queryExecutor, source, period, units);
}
@Override
diff --git a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
index 32305f5..e746baf 100644
--- a/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
+++ b/extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java
@@ -33,11 +33,12 @@
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.rya.streams.api.queries.QueryChangeLog;
import org.apache.rya.streams.kafka.KafkaTopics;
import org.apache.rya.streams.kafka.queries.KafkaQueryChangeLog;
import org.apache.rya.streams.kafka.queries.KafkaQueryChangeLogFactory;
import org.apache.rya.streams.querymanager.QueryChangeLogSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AbstractScheduledService;
@@ -53,6 +54,8 @@
@DefaultAnnotation(NonNull.class)
public class KafkaQueryChangeLogSource extends AbstractScheduledService implements QueryChangeLogSource {
+ private static final Logger log = LoggerFactory.getLogger(KafkaQueryChangeLogSource.class);
+
/**
* Ensures thread safe interactions with this object.
*/
@@ -74,10 +77,10 @@
private final Set<SourceListener> listeners = new HashSet<>();
/**
- * Maps Rya instance name to a Query Change Log for that instance. This map is used to keep
- * track of how the change logs change over time within the Kafka Server.
+ * Maps Rya instance names to the Query Change Log topic name in Kafka. This map is used to
+ * keep track of how the change logs change over time within the Kafka Server.
*/
- private final HashMap<String, QueryChangeLog> knownChangeLogs = new HashMap<>();
+ private final HashMap<String, String> knownChangeLogs = new HashMap<>();
/**
* A consumer that is used to poll the Kafka Server for topics.
@@ -101,6 +104,8 @@
@Override
protected void startUp() throws Exception {
+ log.info("Kafka Query Change Log Source watching " + kafkaBootstrapServer + " starting up...");
+
// Setup the consumer that is used to list topics for the source.
final Properties consumerProperties = new Properties();
consumerProperties.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServer);
@@ -108,17 +113,23 @@
consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
listTopicsConsumer = new KafkaConsumer<>(consumerProperties);
+
+ log.info("Kafka Query Change Log Source watching " + kafkaBootstrapServer + " started.");
}
@Override
protected void shutDown() throws Exception {
- // Shut down the consumer that's used to list topics.
- listTopicsConsumer.close();
+ log.info("Kafka Query Change Log Source watching " + kafkaBootstrapServer + " shutting down...");
- // Shut down all of the change logs that were created within this class.
- for(final QueryChangeLog changeLog : knownChangeLogs.values()) {
- changeLog.close();
+ lock.lock();
+ try {
+ // Shut down the consumer that's used to list topics.
+ listTopicsConsumer.close();
+ } finally {
+ lock.unlock();
}
+
+ log.info("Kafka Query Change Log Source watching " + kafkaBootstrapServer + " shut down.");
}
@Override
@@ -130,8 +141,10 @@
listeners.add(listener);
// Notify it with everything that already exists.
- for(final Entry<String, QueryChangeLog> entry : knownChangeLogs.entrySet()) {
- listener.notifyCreate(entry.getKey(), entry.getValue());
+ for(final Entry<String, String> entry : knownChangeLogs.entrySet()) {
+ final String changeLogTopic = entry.getValue();
+ final KafkaQueryChangeLog changeLog = KafkaQueryChangeLogFactory.make(kafkaBootstrapServer, changeLogTopic);
+ listener.notifyCreate(entry.getKey(), changeLog);
}
} finally {
lock.unlock();
@@ -174,26 +187,23 @@
// Handle the deletes.
for(final String deletedRyaInstance : deletedRyaInstances) {
// Remove the change log from the set of known logs.
- final QueryChangeLog removed = knownChangeLogs.remove(deletedRyaInstance);
+ knownChangeLogs.remove(deletedRyaInstance);
- // Notify the listeners of the update.
+ // Notify the listeners of the update so that they may close the previously provided change log.
for(final SourceListener listener : listeners) {
listener.notifyDelete(deletedRyaInstance);
}
-
- // Ensure the change log is closed.
- removed.close();
}
// Handle the adds.
for(final String createdRyaInstance : createdRyaInstances) {
// Create and store the ChangeLog.
final String changeLogTopic = KafkaTopics.queryChangeLogTopic(createdRyaInstance);
- final KafkaQueryChangeLog changeLog = KafkaQueryChangeLogFactory.make(kafkaBootstrapServer, changeLogTopic);
- knownChangeLogs.put(createdRyaInstance, changeLog);
+ knownChangeLogs.put(createdRyaInstance, changeLogTopic);
// Notify the listeners of the update.
for(final SourceListener listener : listeners) {
+ final KafkaQueryChangeLog changeLog = KafkaQueryChangeLogFactory.make(kafkaBootstrapServer, changeLogTopic);
listener.notifyCreate(createdRyaInstance, changeLog);
}
}
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 947a215..3a59636 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
@@ -32,10 +32,15 @@
import org.apache.rya.streams.api.entity.StreamsQuery;
import org.apache.rya.streams.kafka.KafkaStreamsFactory;
import org.apache.rya.streams.kafka.KafkaStreamsFactory.KafkaStreamsFactoryException;
+import org.apache.rya.streams.kafka.KafkaTopics;
+import org.apache.rya.streams.kafka.interactor.CreateKafkaTopic;
import org.apache.rya.streams.querymanager.QueryExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AbstractIdleService;
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
@@ -50,6 +55,7 @@
*/
@DefaultAnnotation(NonNull.class)
public class LocalQueryExecutor extends AbstractIdleService implements QueryExecutor {
+ private static final Logger log = LoggerFactory.getLogger(LocalQueryExecutor.class);
/**
* Provides thread safety when interacting with this class.
@@ -72,6 +78,11 @@
private final Map<UUID, KafkaStreams> byQueryId = new HashMap<>();
/**
+ * Used to create the input and output topics for a Kafka Streams job.
+ */
+ private final CreateKafkaTopic createKafkaTopic;
+
+ /**
* Builds the {@link KafkaStreams} objects that execute {@link KafkaStream}s.
*/
private final KafkaStreamsFactory streamsFactory;
@@ -79,23 +90,31 @@
/**
* Constructs an instance of {@link LocalQueryExecutor}.
*
+ * @param createKafkaTopic - Used to create the input and output topics for a Kafka Streams job. (not null)
* @param streamsFactory - Builds the {@link KafkaStreams} objects that execute {@link KafkaStream}s. (not null)
*/
- public LocalQueryExecutor(final KafkaStreamsFactory streamsFactory) {
+ public LocalQueryExecutor(
+ final CreateKafkaTopic createKafkaTopic,
+ final KafkaStreamsFactory streamsFactory) {
+ this.createKafkaTopic = requireNonNull(createKafkaTopic);
this.streamsFactory = requireNonNull(streamsFactory);
}
@Override
protected void startUp() throws Exception {
- // Nothing to do.
+ log.info("Local Query Executor starting up.");
}
@Override
protected void shutDown() throws Exception {
+ log.info("Local Query Executor shutting down. Stopping all jobs...");
+
// Stop all of the running queries.
for(final KafkaStreams job : byQueryId.values()) {
job.close();
}
+
+ log.info("Local Query Executor shut down.");
}
@Override
@@ -106,6 +125,14 @@
lock.lock();
try {
+ // Make sure the Statements topic exists for the query.
+ final Set<String> topics = Sets.newHashSet(
+ KafkaTopics.statementsTopic(ryaInstance),
+ KafkaTopics.queryResultsTopic(query.getQueryId()));
+
+ // Make sure the Query Results topic exists for the query.
+ createKafkaTopic.createTopics(topics, 1, 1);
+
// Setup the Kafka Streams job that will execute.
final KafkaStreams streams = streamsFactory.make(ryaInstance, query);
streams.start();
diff --git a/extras/rya.streams/query-manager/src/main/xsd/QueryManagerConfig.xsd b/extras/rya.streams/query-manager/src/main/xsd/QueryManagerConfig.xsd
index c1285d4..21170bb 100644
--- a/extras/rya.streams/query-manager/src/main/xsd/QueryManagerConfig.xsd
+++ b/extras/rya.streams/query-manager/src/main/xsd/QueryManagerConfig.xsd
@@ -30,6 +30,13 @@
</xs:choice>
</xs:complexType>
</xs:element>
+ <xs:element name="queryExecutor">
+ <xs:complexType>
+ <xs:choice>
+ <xs:element name="localKafkaStreams" type="localKafkaStreams"/>
+ </xs:choice>
+ </xs:complexType>
+ </xs:element>
<xs:element name="performanceTunning">
<xs:complexType>
<xs:sequence>
@@ -65,6 +72,13 @@
</xs:sequence>
</xs:complexType>
+ <!-- Define what a local Kafka Streams query executor looks like. -->
+ <xs:complexType name="localKafkaStreams">
+ <xs:sequence>
+ <xs:element name="zookeepers" type="xs:string"/>
+ </xs:sequence>
+ </xs:complexType>
+
<!-- Define the legal range for a TCP port. -->
<xs:simpleType name="tcpPort">
<xs:restriction base="xs:int">
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkGeneratorTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkGeneratorTest.java
new file mode 100644
index 0000000..da9be78
--- /dev/null
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkGeneratorTest.java
@@ -0,0 +1,136 @@
+/**
+ * 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.streams.querymanager;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.streams.api.queries.QueryChangeLog;
+import org.apache.rya.streams.querymanager.QueryManager.LogEvent;
+import org.apache.rya.streams.querymanager.QueryManager.LogEvent.LogEventType;
+import org.apache.rya.streams.querymanager.QueryManager.LogEventWorkGenerator;
+import org.junit.Test;
+
+/**
+ * Unit tests the methods of {@link LogEventWorkGenerator}.
+ */
+public class LogEventWorkGeneratorTest {
+
+ @Test
+ public void shutdownSignalKillsThread() {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<LogEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the LogEventWorkGenerator work.
+ final LogEventWorkGenerator generator = new LogEventWorkGenerator(queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a created change log.
+ final Thread notifyThread = new Thread(() -> {
+ generator.notifyCreate("rya", mock(QueryChangeLog.class));
+ });
+
+ // Fill the queue so that nothing may be offered to it.
+ queue.offer(LogEvent.delete("rya"));
+
+ // Start the thread and show that it is still alive after the offer period.
+ notifyThread.start();
+ assertTrue( ThreadUtil.stillAlive(notifyThread, 200) );
+
+ // Set the shutdown signal to true and join the thread. If we were able to join, then it shut down.
+ shutdownSignal.set(true);
+ assertFalse( ThreadUtil.stillAlive(notifyThread, 1000) );
+ }
+
+ @Test
+ public void notifyCreate() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<LogEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the LogEventWorkGenerator work.
+ final LogEventWorkGenerator generator = new LogEventWorkGenerator(queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a created change log.
+ final CountDownLatch notified = new CountDownLatch(1);
+ final Thread notifyThread = new Thread(() -> {
+ generator.notifyCreate("rya", mock(QueryChangeLog.class));
+ notified.countDown();
+ });
+
+ try {
+ // Start the thread that performs the notification.
+ notifyThread.start();
+
+ // Wait for the thread to indicate it has notified and check the queue for the value.
+ notified.await(200, TimeUnit.MILLISECONDS);
+ final LogEvent event = queue.poll(200, TimeUnit.MILLISECONDS);
+ assertEquals(LogEventType.CREATE, event.getEventType());
+ assertEquals("rya", event.getRyaInstanceName());
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+
+ @Test
+ public void notifyDelete() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<LogEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the LogEventWorkGenerator work.
+ final LogEventWorkGenerator generator = new LogEventWorkGenerator(queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a deleted change log.
+ final CountDownLatch notified = new CountDownLatch(1);
+ final Thread notifyThread = new Thread(() -> {
+ generator.notifyDelete("rya");
+ notified.countDown();
+ });
+
+ try {
+ // Start the thread that performs the notification.
+ notifyThread.start();
+
+ // Wait for the thread to indicate it has notified and check the queue for the value.
+ notified.await(200, TimeUnit.MILLISECONDS);
+ final LogEvent event = queue.poll(200, TimeUnit.MILLISECONDS);
+ assertEquals(LogEventType.DELETE, event.getEventType());
+ assertEquals("rya", event.getRyaInstanceName());
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkerTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkerTest.java
new file mode 100644
index 0000000..cb708ed
--- /dev/null
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/LogEventWorkerTest.java
@@ -0,0 +1,245 @@
+/**
+ * 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.streams.querymanager;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.api.queries.InMemoryQueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChange;
+import org.apache.rya.streams.api.queries.QueryChangeLog;
+import org.apache.rya.streams.querymanager.QueryManager.LogEvent;
+import org.apache.rya.streams.querymanager.QueryManager.LogEventWorker;
+import org.apache.rya.streams.querymanager.QueryManager.QueryEvent;
+import org.junit.Test;
+
+/**
+ * Unit tests the methods of {@link LogEventWorker}.
+ */
+public class LogEventWorkerTest {
+
+ @Test
+ public void shutdownSignalKillsThread() {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The thread that will perform the LogEventWorker task.
+ final Thread logEventWorker = new Thread(new LogEventWorker(new ArrayBlockingQueue<>(1),
+ new ArrayBlockingQueue<>(1), 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ logEventWorker.start();
+
+ // Wait longer than the poll time to see if the thread died. Show that it is still running.
+ assertTrue(ThreadUtil.stillAlive(logEventWorker, 200));
+
+ // Set the shutdown signal to true and join the thread. If we were able to join, then it shut down.
+ shutdownSignal.set(true);
+ assertFalse(ThreadUtil.stillAlive(logEventWorker, 500));
+ }
+
+ @Test
+ public void nofity_logCreated_doesNotExist() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to feed work.
+ final BlockingQueue<LogEvent> logEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The queue work is written to.
+ final BlockingQueue<QueryEvent> queryEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The Query Change Log that will be watched.
+ final QueryChangeLog changeLog = new InMemoryQueryChangeLog();
+
+ // Write a message that indicates a new query should be active.
+ final UUID firstQueryId = UUID.randomUUID();
+ changeLog.write(QueryChange.create(firstQueryId, "select * where { ?a ?b ?c . }", true));
+
+ // Write a message that adds an active query, but then makes it inactive. Because both of these
+ // events are written to the log before the worker subscribes to the repository for updates, they
+ // must result in a single query stopped event.
+ final UUID secondQueryId = UUID.randomUUID();
+ changeLog.write(QueryChange.create(secondQueryId, "select * where { ?d ?e ?f . }", true));
+ changeLog.write(QueryChange.update(secondQueryId, false));
+
+ // Start the worker that will be tested.
+ final Thread logEventWorker = new Thread(new LogEventWorker(logEventQueue,
+ queryEventQueue, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ logEventWorker.start();
+
+ try {
+ // Write a unit of work that indicates a log was created.
+ final LogEvent createLogEvent = LogEvent.create("rya", changeLog);
+ logEventQueue.offer(createLogEvent);
+
+ // We must see the following Query Events added to the work queue.
+ // Query 1, executing.
+ // Query 2, stopped.
+ Set<QueryEvent> expectedEvents = new HashSet<>();
+ expectedEvents.add(QueryEvent.executing("rya",
+ new StreamsQuery(firstQueryId, "select * where { ?a ?b ?c . }", true)));
+ expectedEvents.add(QueryEvent.stopped("rya", secondQueryId));
+
+ Set<QueryEvent> queryEvents = new HashSet<>();
+ queryEvents.add( queryEventQueue.poll(500, TimeUnit.MILLISECONDS) );
+ queryEvents.add( queryEventQueue.poll(500, TimeUnit.MILLISECONDS) );
+
+ assertEquals(expectedEvents, queryEvents);
+
+ // Write an event to the change log that stops the first query.
+ changeLog.write(QueryChange.update(firstQueryId, false));
+
+ // Show it was also reflected in the changes.
+ // Query 1, stopped.
+ expectedEvents = new HashSet<>();
+ expectedEvents.add(QueryEvent.stopped("rya", firstQueryId));
+
+ queryEvents = new HashSet<>();
+ queryEvents.add( queryEventQueue.poll(500, TimeUnit.MILLISECONDS) );
+
+ assertEquals(expectedEvents, queryEvents);
+ } finally {
+ shutdownSignal.set(true);
+ logEventWorker.join();
+ }
+ }
+
+ @Test
+ public void nofity_logCreated_exists() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to feed work.
+ final BlockingQueue<LogEvent> logEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The queue work is written to.
+ final BlockingQueue<QueryEvent> queryEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The Query Change Log that will be watched.
+ final QueryChangeLog changeLog = new InMemoryQueryChangeLog();
+
+ // Write a message that indicates a new query should be active.
+ final UUID firstQueryId = UUID.randomUUID();
+ changeLog.write(QueryChange.create(firstQueryId, "select * where { ?a ?b ?c . }", true));
+
+ // Start the worker that will be tested.
+ final Thread logEventWorker = new Thread(new LogEventWorker(logEventQueue,
+ queryEventQueue, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ logEventWorker.start();
+
+ try {
+ // Write a unit of work that indicates a log was created.
+ final LogEvent createLogEvent = LogEvent.create("rya", changeLog);
+ logEventQueue.offer(createLogEvent);
+
+ // Say the same log was created a second time.
+ logEventQueue.offer(createLogEvent);
+
+ // Show that only a single unit of work was added for the log. This indicates the
+ // second message was effectively skipped as it would have add its work added twice otherwise.
+ final Set<QueryEvent> expectedEvents = new HashSet<>();
+ expectedEvents.add(QueryEvent.executing("rya",
+ new StreamsQuery(firstQueryId, "select * where { ?a ?b ?c . }", true)));
+
+ final Set<QueryEvent> queryEvents = new HashSet<>();
+ queryEvents.add( queryEventQueue.poll(500, TimeUnit.MILLISECONDS) );
+
+ assertNull(queryEventQueue.poll(500, TimeUnit.MILLISECONDS));
+ assertEquals(expectedEvents, queryEvents);
+ } finally {
+ shutdownSignal.set(true);
+ logEventWorker.join();
+ }
+ }
+
+ @Test
+ public void notify_logDeleted_exists() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to feed work.
+ final BlockingQueue<LogEvent> logEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The queue work is written to.
+ final BlockingQueue<QueryEvent> queryEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The Query Change Log that will be watched.
+ final QueryChangeLog changeLog = new InMemoryQueryChangeLog();
+
+ // Start the worker that will be tested.
+ final Thread logEventWorker = new Thread(new LogEventWorker(logEventQueue,
+ queryEventQueue, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ logEventWorker.start();
+
+ try {
+ // Write a unit of work that indicates a log was created.
+ final LogEvent createLogEvent = LogEvent.create("rya", changeLog);
+ logEventQueue.offer(createLogEvent);
+
+ // Write a unit of work that indicates a log was deleted.
+ logEventQueue.offer(LogEvent.delete("rya"));
+
+ // Show that a single unit of work was created for deleting everything for "rya".
+ assertEquals(QueryEvent.stopALL("rya"), queryEventQueue.poll(500, TimeUnit.MILLISECONDS));
+ assertNull(queryEventQueue.poll(500, TimeUnit.MILLISECONDS));
+ } finally {
+ shutdownSignal.set(true);
+ logEventWorker.join();
+ }
+ }
+
+ @Test
+ public void notify_logDeleted_doesNotExist() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to feed work.
+ final BlockingQueue<LogEvent> logEventQueue = new ArrayBlockingQueue<>(10);
+
+ // The queue work is written to.
+ final BlockingQueue<QueryEvent> queryEventQueue = new ArrayBlockingQueue<>(10);
+
+ // Start the worker that will be tested.
+ final Thread logEventWorker = new Thread(new LogEventWorker(logEventQueue,
+ queryEventQueue, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ logEventWorker.start();
+
+ try {
+ // Write a unit of work that indicates a log was deleted. Since it was never created,
+ // this will not cause anything to be written to the QueryEvent queue.
+ logEventQueue.offer(LogEvent.delete("rya"));
+
+ // Show that a single unit of work was created for deleting everything for "rya".
+ assertNull(queryEventQueue.poll(500, TimeUnit.MILLISECONDS));
+ } finally {
+ shutdownSignal.set(true);
+ logEventWorker.join();
+ }
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkGeneratorTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkGeneratorTest.java
new file mode 100644
index 0000000..4495e19
--- /dev/null
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkGeneratorTest.java
@@ -0,0 +1,265 @@
+/**
+ * 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.streams.querymanager;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.api.queries.ChangeLogEntry;
+import org.apache.rya.streams.api.queries.QueryChange;
+import org.apache.rya.streams.querymanager.QueryManager.QueryEvent;
+import org.apache.rya.streams.querymanager.QueryManager.QueryEventWorkGenerator;
+import org.junit.Test;
+
+/**
+ * Unit tests the methods of {@link QueryEventWorkGenerator}.
+ */
+public class QueryEventWorkGeneratorTest {
+
+ @Test
+ public void shutdownSignalKillsThread() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", new CountDownLatch(1), queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a created query.
+ final Thread notifyThread = new Thread(() -> {
+ generator.notify(mock(ChangeLogEntry.class), Optional.empty());
+ });
+
+ // Fill the queue so that nothing may be offered to it.
+ queue.offer(QueryEvent.stopALL("rya"));
+
+ // Start the thread and show that it is still alive after the offer period.
+ notifyThread.start();
+ assertTrue( ThreadUtil.stillAlive(notifyThread, 200) );
+
+ // Set the shutdown signal to true and join the thread. If we were able to join, then it shut down.
+ shutdownSignal.set(true);
+ assertFalse( ThreadUtil.stillAlive(notifyThread, 1000) );
+ }
+
+ @Test
+ public void waitsForSubscriptionWork() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final CountDownLatch latch = new CountDownLatch(1);
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", latch, queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a created query.
+ final UUID queryId = UUID.randomUUID();
+ final StreamsQuery query = new StreamsQuery(queryId, "query", true);
+ final Thread notifyThread = new Thread(() -> {
+ final QueryChange change = QueryChange.create(queryId, query.getSparql(), query.isActive());
+ final ChangeLogEntry<QueryChange> entry = new ChangeLogEntry<>(0, change);
+ generator.notify(entry, Optional.of(query));
+ });
+
+ // Start the thread.
+ notifyThread.start();
+
+ try {
+ // Wait longer than the blocking period and show the thread is still alive and nothing has been added
+ // to the work queue.
+ Thread.sleep(150);
+ assertTrue( notifyThread.isAlive() );
+
+ // Count down the latch.
+ latch.countDown();
+
+ // Show work was added to the queue and the notifying thread died.
+ final QueryEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
+ final QueryEvent expected = QueryEvent.executing("rya", new StreamsQuery(queryId, query.getSparql(), query.isActive()));
+ assertEquals(expected, event);
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+
+ @Test
+ public void notifyCreate() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final CountDownLatch latch = new CountDownLatch(1);
+ latch.countDown();
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", latch, queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a created query.
+ final UUID queryId = UUID.randomUUID();
+ final StreamsQuery query = new StreamsQuery(queryId, "query", true);
+ final Thread notifyThread = new Thread(() -> {
+ final QueryChange change = QueryChange.create(queryId, query.getSparql(), query.isActive());
+ final ChangeLogEntry<QueryChange> entry = new ChangeLogEntry<>(0, change);
+ generator.notify(entry, Optional.of(query));
+ });
+
+ // Start the thread.
+ notifyThread.start();
+
+ try {
+ // Show work was added to the queue and the notifying thread died.
+ final QueryEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
+ final QueryEvent expected = QueryEvent.executing("rya", new StreamsQuery(queryId, query.getSparql(), query.isActive()));
+ assertEquals(expected, event);
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+
+ @Test
+ public void notifyDelete() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final CountDownLatch latch = new CountDownLatch(1);
+ latch.countDown();
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", latch, queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with a deleted query.
+ final UUID queryId = UUID.randomUUID();
+ final Thread notifyThread = new Thread(() -> {
+ final QueryChange change = QueryChange.delete(queryId);
+ final ChangeLogEntry<QueryChange> entry = new ChangeLogEntry<>(0, change);
+ generator.notify(entry, Optional.empty());
+ });
+
+ // Start the thread.
+ notifyThread.start();
+
+ try {
+ // Show work was added to the queue and the notifying thread died.
+ final QueryEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
+ final QueryEvent expected = QueryEvent.stopped("rya", queryId);
+ assertEquals(expected, event);
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+
+ @Test
+ public void notifyUpdate_isActive() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final CountDownLatch latch = new CountDownLatch(1);
+ latch.countDown();
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", latch, queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with an update query change.
+ final UUID queryId = UUID.randomUUID();
+ final StreamsQuery query = new StreamsQuery(queryId, "query", true);
+ final Thread notifyThread = new Thread(() -> {
+ final QueryChange change = QueryChange.update(queryId, true);
+ final ChangeLogEntry<QueryChange> entry = new ChangeLogEntry<>(0, change);
+ generator.notify(entry, Optional.of(query));
+ });
+
+ // Start the thread.
+ notifyThread.start();
+
+ try {
+ // Show work was added to the queue and the notifying thread died.
+ final QueryEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
+ final QueryEvent expected = QueryEvent.executing("rya", new StreamsQuery(queryId, query.getSparql(), query.isActive()));
+ assertEquals(expected, event);
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+
+ @Test
+ public void notifyUpdate_isNotActive() throws Exception {
+ // The signal that will kill the notifying thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue generated work is offered to.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The listener that will perform the QueryEventWorkGenerator work.
+ final CountDownLatch latch = new CountDownLatch(1);
+ latch.countDown();
+ final QueryEventWorkGenerator generator =
+ new QueryEventWorkGenerator("rya", latch, queue, 50, TimeUnit.MILLISECONDS, shutdownSignal);
+
+ // A thread that will attempt to notify the generator with an update query change.
+ final UUID queryId = UUID.randomUUID();
+ final StreamsQuery query = new StreamsQuery(queryId, "query", false);
+ final Thread notifyThread = new Thread(() -> {
+ final QueryChange change = QueryChange.update(queryId, false);
+ final ChangeLogEntry<QueryChange> entry = new ChangeLogEntry<>(0, change);
+ generator.notify(entry, Optional.of(query));
+ });
+
+ // Start the thread.
+ notifyThread.start();
+
+ try {
+ // Show work was added to the queue and the notifying thread died.
+ final QueryEvent event = queue.poll(500, TimeUnit.MILLISECONDS);
+ final QueryEvent expected = QueryEvent.stopped("rya", queryId);
+ assertEquals(expected, event);
+ } finally {
+ shutdownSignal.set(true);
+ notifyThread.join();
+ }
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkerTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkerTest.java
new file mode 100644
index 0000000..33c0719
--- /dev/null
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryEventWorkerTest.java
@@ -0,0 +1,172 @@
+/**
+ * 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.streams.querymanager;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.querymanager.QueryManager.QueryEvent;
+import org.apache.rya.streams.querymanager.QueryManager.QueryEventWorker;
+import org.junit.Test;
+
+/**
+ * Unit tests the methods of {@link QueryManager.QueryEventWorker}.
+ */
+public class QueryEventWorkerTest {
+
+ @Test
+ public void shutdownSignalKillsThread() {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The thread that will perform the QueryEventWorker task.
+ final Thread queryEventWorker = new Thread(new QueryEventWorker(new ArrayBlockingQueue<>(1),
+ mock(QueryExecutor.class), 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ queryEventWorker.start();
+
+ // Wait longer than the poll time to see if the thread died. Show that it is still running.
+ assertTrue(ThreadUtil.stillAlive(queryEventWorker, 200));
+
+ // Set the shutdown signal to true and join the thread. If we were able to join, then it shut down.
+ shutdownSignal.set(true);
+ assertFalse(ThreadUtil.stillAlive(queryEventWorker, 1000));
+ }
+
+ @Test
+ public void executingWork() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to send the execute work to the thread.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The message that indicates a query needs to be executed.
+ final String ryaInstance = "rya";
+ final StreamsQuery query = new StreamsQuery(UUID.randomUUID(), "sparql", true);
+ final QueryEvent executingEvent = QueryEvent.executing(ryaInstance, query);
+
+ // Release a latch if the startQuery method on the queryExecutor is invoked with the correct values.
+ final CountDownLatch startQueryInvoked = new CountDownLatch(1);
+ final QueryExecutor queryExecutor = mock(QueryExecutor.class);
+ doAnswer(invocation -> {
+ startQueryInvoked.countDown();
+ return null;
+ }).when(queryExecutor).startQuery(ryaInstance, query);
+
+ // The thread that will perform the QueryEventWorker task.
+ final Thread queryEventWorker = new Thread(new QueryEventWorker(queue,
+ queryExecutor, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ try {
+ queryEventWorker.start();
+
+ // Provide a message indicating a query needs to be executing.
+ queue.put(executingEvent);
+
+ // Verify the Query Executor was told to start the query.
+ assertTrue( startQueryInvoked.await(150, TimeUnit.MILLISECONDS) );
+ } finally {
+ shutdownSignal.set(true);
+ queryEventWorker.join();
+ }
+ }
+
+ @Test
+ public void stoppedWork() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to send the execute work to the thread.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The message that indicates a query needs to be stopped.
+ final UUID queryId = UUID.randomUUID();
+ final QueryEvent stoppedEvent = QueryEvent.stopped("rya", queryId);
+
+ // Release a latch if the stopQuery method on the queryExecutor is invoked with the correct values.
+ final CountDownLatch stopQueryInvoked = new CountDownLatch(1);
+ final QueryExecutor queryExecutor = mock(QueryExecutor.class);
+ doAnswer(invocation -> {
+ stopQueryInvoked.countDown();
+ return null;
+ }).when(queryExecutor).stopQuery(queryId);
+
+ final Thread queryEventWorker = new Thread(new QueryEventWorker(queue,
+ queryExecutor, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ try {
+ // The thread that will perform the QueryEventWorker task.
+ queryEventWorker.start();
+
+ // Provide a message indicating a query needs to be executing.
+ queue.put(stoppedEvent);
+
+ // Verify the Query Executor was told to stop the query.
+ assertTrue( stopQueryInvoked.await(150, TimeUnit.MILLISECONDS) );
+ } finally {
+ shutdownSignal.set(true);
+ queryEventWorker.join();
+ }
+ }
+
+ @Test
+ public void stopAllWork() throws Exception {
+ // The signal that will kill the working thread.
+ final AtomicBoolean shutdownSignal = new AtomicBoolean(false);
+
+ // The queue used to send the execute work to the thread.
+ final BlockingQueue<QueryEvent> queue = new ArrayBlockingQueue<>(1);
+
+ // The message that indicates all queries for a rya instance need to be stopped.
+ final String ryaInstance = "rya";
+ final QueryEvent stopAllEvent = QueryEvent.stopALL(ryaInstance);
+
+ // Release a latch if the stopQuery method on the queryExecutor is invoked with the correct values.
+ final CountDownLatch testMethodInvoked = new CountDownLatch(1);
+ final QueryExecutor queryExecutor = mock(QueryExecutor.class);
+ doAnswer(invocation -> {
+ testMethodInvoked.countDown();
+ return null;
+ }).when(queryExecutor).stopAll(ryaInstance);
+
+ final Thread queryEventWorker = new Thread(new QueryEventWorker(queue,
+ queryExecutor, 50, TimeUnit.MILLISECONDS, shutdownSignal));
+ try {
+ // The thread that will perform the QueryEventWorker task.
+ queryEventWorker.start();
+
+ // Provide a message indicating a query needs to be executing.
+ queue.put(stopAllEvent);
+
+ // Verify the Query Executor was told to stop all the queries.
+ assertTrue( testMethodInvoked.await(150, TimeUnit.MILLISECONDS) );
+ } finally {
+ shutdownSignal.set(true);
+ queryEventWorker.join();
+ }
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryManagerTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryManagerTest.java
index a1203a0..04e70c0 100644
--- a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryManagerTest.java
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/QueryManagerTest.java
@@ -1,18 +1,20 @@
-/*
- * 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
+/**
+ * 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
+ * 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.
+ * 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.streams.querymanager;
@@ -34,13 +36,10 @@
import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
import org.junit.Test;
-import com.google.common.util.concurrent.AbstractScheduledService.Scheduler;
-
/**
- * Test for the {@link QueryManager}
+ * Unit tests the methods of {@link QueryManager}.
*/
public class QueryManagerTest {
- private static final Scheduler TEST_SCHEDULER = Scheduler.newFixedRateSchedule(0, 100, TimeUnit.MILLISECONDS);
/**
* Tests when the query manager is notified to create a new query, the query
@@ -74,7 +73,7 @@
return null;
}).when(source).subscribe(any(SourceListener.class));
- final QueryManager qm = new QueryManager(qe, source, TEST_SCHEDULER);
+ final QueryManager qm = new QueryManager(qe, source, 50, TimeUnit.MILLISECONDS);
try {
qm.startAndWait();
queryStarted.await(5, TimeUnit.SECONDS);
@@ -128,7 +127,7 @@
return null;
}).when(source).subscribe(any(SourceListener.class));
- final QueryManager qm = new QueryManager(qe, source, TEST_SCHEDULER);
+ final QueryManager qm = new QueryManager(qe, source, 50, TimeUnit.MILLISECONDS);
try {
qm.startAndWait();
queryDeleted.await(5, TimeUnit.SECONDS);
@@ -183,7 +182,7 @@
return null;
}).when(source).subscribe(any(SourceListener.class));
- final QueryManager qm = new QueryManager(qe, source, TEST_SCHEDULER);
+ final QueryManager qm = new QueryManager(qe, source, 50, TimeUnit.MILLISECONDS);
try {
qm.startAndWait();
queryDeleted.await(10, TimeUnit.SECONDS);
@@ -192,4 +191,4 @@
qm.stopAndWait();
}
}
-}
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/ThreadUtil.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/ThreadUtil.java
new file mode 100644
index 0000000..9896e31
--- /dev/null
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/ThreadUtil.java
@@ -0,0 +1,48 @@
+/**
+ * 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.streams.querymanager;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Utilities that are useful for interacting with {@link Thread}s while testing.
+ */
+public class ThreadUtil {
+
+ /**
+ * Private constructor to prevent instantiation.
+ */
+ private ThreadUtil() { }
+
+ /**
+ * A utility function that returns whether a thread is alive or not after waiting
+ * some specified period of time to join it.
+ *
+ * @param thread - The thread that will be joined. (not null)
+ * @param millis - How long to wait to join the thread.
+ * @return {@code true} if the thread is still alive, otherwise {@code false}.
+ */
+ public static boolean stillAlive(final Thread thread, final long millis) {
+ requireNonNull(thread);
+ try {
+ thread.join(millis);
+ } catch (final InterruptedException e) { }
+ return thread.isAlive();
+ }
+}
\ No newline at end of file
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
index 3cbe894..f9c8a03 100644
--- a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorIT.java
@@ -35,6 +35,7 @@
import org.apache.rya.streams.kafka.KafkaStreamsFactory;
import org.apache.rya.streams.kafka.KafkaTopics;
import org.apache.rya.streams.kafka.SingleThreadKafkaStreamsFactory;
+import org.apache.rya.streams.kafka.interactor.CreateKafkaTopic;
import org.apache.rya.streams.kafka.interactor.KafkaLoadStatements;
import org.apache.rya.streams.kafka.serialization.VisibilityBindingSetDeserializer;
import org.apache.rya.streams.kafka.serialization.VisibilityStatementSerializer;
@@ -119,9 +120,10 @@
expected.add(new VisibilityBindingSet(bs, "a"));
// Start the executor that will be tested.
+ final CreateKafkaTopic createKafkaTopic = new CreateKafkaTopic( kafka.getZookeeperServers() );
final String kafkaServers = kafka.getKafkaHostname() + ":" + kafka.getKafkaPort();
final KafkaStreamsFactory jobFactory = new SingleThreadKafkaStreamsFactory(kafkaServers);
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(createKafkaTopic, jobFactory);
executor.startAndWait();
try {
// Start the query.
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorTest.java
index 0df5794..c0f888e 100644
--- a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorTest.java
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/kafka/LocalQueryExecutorTest.java
@@ -32,6 +32,7 @@
import org.apache.kafka.streams.KafkaStreams;
import org.apache.rya.streams.api.entity.StreamsQuery;
import org.apache.rya.streams.kafka.KafkaStreamsFactory;
+import org.apache.rya.streams.kafka.interactor.CreateKafkaTopic;
import org.apache.rya.streams.querymanager.QueryExecutor;
import org.junit.Test;
@@ -44,7 +45,7 @@
@Test(expected = IllegalStateException.class)
public void startQuery_serviceNotStarted() throws Exception {
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.startQuery("rya", new StreamsQuery(UUID.randomUUID(), "query", true));
}
@@ -60,7 +61,7 @@
when(jobFactory.make(eq(ryaInstance), eq(query))).thenReturn(queryJob);
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Tell the executor to start the query.
@@ -75,14 +76,14 @@
@Test(expected = IllegalStateException.class)
public void stopQuery_serviceNotStarted() throws Exception {
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.stopQuery(UUID.randomUUID());
}
@Test
public void stopQuery_queryNotRunning() throws Exception {
// Start an executor.
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.startAndWait();
try {
// Try to stop a query that was never stareted.
@@ -104,7 +105,7 @@
when(jobFactory.make(eq(ryaInstance), eq(query))).thenReturn(queryJob);
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Tell the executor to start the query.
@@ -122,7 +123,7 @@
@Test(expected = IllegalStateException.class)
public void stopAll_serviceNotStarted() throws Exception {
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.stopAll("rya");
}
@@ -141,7 +142,7 @@
when(jobFactory.make(eq(ryaInstance), eq(query2))).thenReturn(queryJob2);
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Tell the executor to start the queries.
@@ -180,7 +181,7 @@
when(jobFactory.make(eq(ryaInstance2), eq(query2))).thenReturn(queryJob2);
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Tell the executor to start the queries.
@@ -205,14 +206,14 @@
@Test(expected = IllegalStateException.class)
public void getRunningQueryIds_serviceNotStarted() throws Exception {
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.getRunningQueryIds();
}
@Test
public void getRunningQueryIds_noneStarted() throws Exception {
// Start an executor.
- final QueryExecutor executor = new LocalQueryExecutor(mock(KafkaStreamsFactory.class));
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), mock(KafkaStreamsFactory.class));
executor.startAndWait();
try {
// Get the list of running queries.
@@ -240,7 +241,7 @@
when(jobFactory.make(eq(ryaInstance), eq(query3))).thenReturn(mock(KafkaStreams.class));
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Start the queries.
@@ -275,7 +276,7 @@
when(jobFactory.make(eq(ryaInstance), eq(query3))).thenReturn(mock(KafkaStreams.class));
// Start the executor that will be tested.
- final QueryExecutor executor = new LocalQueryExecutor(jobFactory);
+ final QueryExecutor executor = new LocalQueryExecutor(mock(CreateKafkaTopic.class), jobFactory);
executor.startAndWait();
try {
// Start the queries.
diff --git a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/xml/QueryManagerConfigMarshallerTest.java b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/xml/QueryManagerConfigMarshallerTest.java
index f2b50ab..de6b9f3 100644
--- a/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/xml/QueryManagerConfigMarshallerTest.java
+++ b/extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/xml/QueryManagerConfigMarshallerTest.java
@@ -45,6 +45,11 @@
" <port>6</port>\n" +
" </kafka>\n" +
" </queryChangeLogSource>\n" +
+ " <queryExecutor>\n" +
+ " <localKafkaStreams>\n" +
+ " <zookeepers>zoo1,zoo2,zoo3</zookeepers>\n" +
+ " </localKafkaStreams>\n" +
+ " </queryExecutor>\n" +
" <performanceTunning>\n" +
" <queryChanngeLogDiscoveryPeriod>\n" +
" <value>1</value>\n" +
