You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@rya.apache.org by kchilton2 <gi...@git.apache.org> on 2018/02/08 21:15:50 UTC
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
GitHub user kchilton2 opened a pull request:
https://github.com/apache/incubator-rya/pull/272
RYA-443 Rya Streams Query Manager daemon program.
## Description
Implemented an RPM installed CentOS 7 daemon service used to execute Kafka Streams jobs on the local machine. This is part of Rya Streams.
### Tests
Wrote automated tests and executed the application on a CentOS 7 machine using the RPM.
### Links
https://issues.apache.org/jira/browse/RYA-443
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kchilton2/incubator-rya RYA-443_rya-streams-querymanager
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-rya/pull/272.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #272
----
commit 6688c2ece024132d3dc6903af9d1003079d330eb
Author: Andrew Smith <sm...@...>
Date: 2018-01-16T17:39:34Z
RYA-446 Daemonifying Query Manager
commit 8803d7d99f8c7189991be181dd11bd0fb9f1e704
Author: Andrew Smith <sm...@...>
Date: 2018-01-18T22:51:40Z
RYA-449 Create QueryChangeLogSource
commit c048db73a5211a18ab649a2647dfa7f7fb929166
Author: Andrew Smith <sm...@...>
Date: 2018-01-19T20:43:55Z
Rya 454 added QueryExecutor interface
commit 0dad1290d593065561fae079b837fcbc479795ed
Author: Andrew Smith <sm...@...>
Date: 2018-01-23T20:20:50Z
Rya 452 Updated QueryRepository
Updated QueryRepository to be a Service
Updated InMemoryQueryRepository to be an AbstractScheduledService
Added listeners to InMemoryQueryRepository
commit 3810fb709a4a97ce729e93733edd9af10d6e5445
Author: Andrew Smith <sm...@...>
Date: 2018-01-23T20:44:32Z
RYA-455 stopAll queries for a rya instance
commit e774ef02161686d9c2f0c6e99ea3b8a438fd240a
Author: kchilton2 <ke...@...>
Date: 2018-01-23T20:50:17Z
RYA-448 Implement JAXB marshalling code for the Query Manager's XML configuration file.
commit 13cfd43ba3609c0690aeebe27eb7b76a88d2ddbd
Author: kchilton2 <ke...@...>
Date: 2018-01-23T21:17:06Z
RYA-450 Implemented a Kafka backed QueryChangeLogSource.
commit eee8d830abc95e4082d906fe7e0dad5944d96b87
Author: kchilton2 <ke...@...>
Date: 2018-01-26T17:06:07Z
RYA-458 Updating the configuration XML so that you may only specify a single QueryChangeLogSource.
commit 815d995c359ae15b5669c1e7c7e375d7c67f8dec
Author: kchilton2 <ke...@...>
Date: 2018-01-26T20:55:59Z
RYA-456 Implement a Single Node implementation of QueryExecutor.
commit ba2898a8c4ba58ad3e38323f28389f75ab262c83
Author: Andrew Smith <sm...@...>
Date: 2018-01-30T19:01:54Z
Rya 451 Query manager
QueryManager with tests
updated InMemoryQueryRepository and its tests
commit 34e616ce63ac4b93a64ca11a10cde148760afc52
Author: kchilton2 <ke...@...>
Date: 2018-01-30T19:44:58Z
RYA-453 Implement the Query Manager's Daemon that controls the application.
commit 8d35b4541bd39734cc68c7e870027760e537d1e7
Author: kchilton2 <ke...@...>
Date: 2018-01-25T22:19:58Z
RYA-446 Create a bin and rpm distribution for the Rya Streams Query Manager application.
commit 4f4e0f4ce0d2eb2d8651b8f569a8f59c8eaaf571
Author: Andrew Smith <sm...@...>
Date: 2018-02-01T21:04:48Z
RYA-446 Service unit file for systemd
commit 4a4456e7cfd18fdd9044b67152252b3623ebe466
Author: kchilton2 <ke...@...>
Date: 2018-02-01T23:31:41Z
RYA-446 Making the Rya Streams Query Manager run as a service on CentOS 7.
commit b889bc0d60e3884ea31edbcb12fcc21331f77963
Author: kchilton2 <ke...@...>
Date: 2018-02-03T03:47:59Z
RYA-451 Fixing threading issues with the QueryManager class.
----
---
[GitHub] incubator-rya issue #272: RYA-443 Rya Streams Query Manager daemon program.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit commented on the issue:
https://github.com/apache/incubator-rya/pull/272
Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/691/<h2>Build result: FAILURE</span></h2>[...truncated 3.03 MB...][INFO] Apache Rya Spark Support ........................... SKIPPED[INFO] Apache Rya Web Projects ............................ SKIPPED[INFO] Apache Rya Web Implementation ...................... SKIPPED[INFO] ------------------------------------------------------------------------[INFO] BUILD FAILURE[INFO] ------------------------------------------------------------------------[INFO] Total time: 37:36 min[INFO] Finished at: 2018-02-14T20:56:07+00:00[INFO] Final Memory: 535M/2904M[INFO] ------------------------------------------------------------------------[ERROR] Failed to execute goal org.codehaus.mojo:rpm-maven-plugin:2.1.5:attached-rpm (create-rpm-distribution) on project rya.streams.query-manager: Unable to query for default vendor from RPM: Error while executing process. Cannot run program "rpm": error=2, No such file or directory
-> [Help 1][ERROR] [ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.[ERROR] Re-run Maven using the -X switch to enable full debug logging.[ERROR] [ERROR] For more information about the errors and possible solutions, please read the following articles:[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException[ERROR] [ERROR] After correcting the problems, you can resume the build with the command[ERROR] mvn <goals> -rf :rya.streams.query-managerchannel stoppedSetting status of 6428ca81a5cb0ef5bd2e762452f1a6e46dff88fc to FAILURE with url https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/691/ and message: 'FAILURE 'Using context: Jenkins: clean package -Pgeoindexing
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276853
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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")
--- End diff --
I'm going to assume it has been sanitized by this point.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167241001
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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")
--- End diff --
Add a space after the semi-colon in "Rya Instance:"
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168278274
--- Diff: extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java ---
@@ -107,12 +112,11 @@ private String formatQueries(final Set<StreamsQuery> queries) {
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( query.isActive() ? " " : " " )
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r170073123
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java ---
@@ -0,0 +1,219 @@
+/**
+ * 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.kafka;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+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.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;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Represents a Kafka Server that manages {@link KafkaQueryChangeLog}s.
+ * <p/>
+ * Thread safe.
--- End diff --
net.jcip.annotations.ThreadSafe
Looks like Rya is using this one:
```
<dependency>
<groupId>com.github.stephenc.jcip</groupId>
<artifactId>jcip-annotations</artifactId>
<version>1.0-1</version>
</dependency>
```
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167242987
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.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 java.util.Objects.requireNonNull;
+
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
+
+import javax.xml.bind.JAXBException;
+
+import org.apache.commons.daemon.Daemon;
+import org.apache.commons.daemon.DaemonContext;
+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;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfig;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfig.PerformanceTunning.QueryChanngeLogDiscoveryPeriod;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfigUnmarshaller;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xml.sax.SAXException;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.google.common.util.concurrent.AbstractScheduledService.Scheduler;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * JSVC integration code for a {@link QueryManager} to be used as a non-Windows daemon.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManagerDaemon implements Daemon {
+
+ private static final Logger log = LoggerFactory.getLogger(QueryManagerDaemon.class);
+
+ /**
+ * The default configuration file's path for the application.
+ */
+ private static final Path DEFAULT_CONFIGURATION_PATH = Paths.get("config/configuration.xml");
+
+ /**
+ * Command line parameters that are used by all commands that interact with Kafka.
+ */
+ class DaemonParameters {
+ @Parameter(names = {"--config", "-c"}, required = false, description = "The path to the application's configuration file.")
+ public String config;
+ }
+
+ private QueryManager manager = null;
+
+ @Override
+ public void init(final DaemonContext context) throws DaemonInitException, Exception {
+ requireNonNull(context);
+
+ // Parse the command line arguments for the configuration file to use.
+ final String[] args = context.getArguments();
+ final DaemonParameters params = new DaemonParameters();
+ try {
+ new JCommander(params).parse(args);
+ } catch(final ParameterException e) {
+ throw new DaemonInitException("Unable to parse the command line arguments.", e);
+ }
+ final Path configFile = params.config != null ? Paths.get(params.config) : DEFAULT_CONFIGURATION_PATH;
+ log.info("Loading the following configuration file: " + configFile);
+
+ // Unmarshall the configuration file into an object.
+ final QueryManagerConfig config;
+ try(InputStream stream = Files.newInputStream(configFile)) {
--- End diff --
final
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167253121
--- Diff: extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java ---
@@ -107,12 +112,11 @@ private String formatQueries(final Set<StreamsQuery> queries) {
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( query.isActive() ? " " : " " )
+ .append("Query: ").append(query.getSparql()).append("\n");
--- End diff --
LogUtils.clean(query.getSparql())
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168274700
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java ---
@@ -243,8 +274,68 @@ private void updateCache() {
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");
--- End diff --
Yep, done.
---
[GitHub] incubator-rya issue #272: RYA-443 Rya Streams Query Manager daemon program.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit commented on the issue:
https://github.com/apache/incubator-rya/pull/272
Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/686/<h2>Build result: FAILURE</span></h2>[...truncated 3.04 MB...][INFO] Apache Rya Web Projects ............................ SKIPPED[INFO] Apache Rya Web Implementation ...................... SKIPPED[INFO] ------------------------------------------------------------------------[INFO] BUILD FAILURE[INFO] ------------------------------------------------------------------------[INFO] Total time: 38:10 min[INFO] Finished at: 2018-02-08T22:18:44+00:00[INFO] Final Memory: 301M/2452M[INFO] ------------------------------------------------------------------------Waiting for Jenkins to finish collecting data[ERROR] Failed to execute goal org.codehaus.mojo:rpm-maven-plugin:2.1.5:attached-rpm (create-rpm-distribution) on project rya.streams.query-manager: Unable to query for default vendor from RPM: Error while executing process. Cannot run program "rpm": error=2, No such file or directory -> [Help 1][ERROR] [ER
ROR] To see the full stack trace of the errors, re-run Maven with the -e switch.[ERROR] Re-run Maven using the -X switch to enable full debug logging.[ERROR] [ERROR] For more information about the errors and possible solutions, please read the following articles:[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException[ERROR] [ERROR] After correcting the problems, you can resume the build with the command[ERROR] mvn <goals> -rf :rya.streams.query-managerchannel stoppedSetting status of b889bc0d60e3884ea31edbcb12fcc21331f77963 to FAILURE with url https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/686/ and message: 'FAILURE 'Using context: Jenkins: clean package -Pgeoindexing
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276101
--- Diff: extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java ---
@@ -48,7 +51,28 @@
*/
public static String queryChangeLogTopic(final String ryaInstance) {
requireNonNull(ryaInstance);
- return ryaInstance + "-QueryChangeLog";
+ return ryaInstance + QUERY_CHANGE_LOG_TOPIC_SUFFIX;
+ }
+
+ /**
+ * Get the Rya instance name from a Kafka topic name that has been used for a {@link QueryChnageLog}.
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276546
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
--- End diff --
Done.
---
[GitHub] incubator-rya issue #272: RYA-443 Rya Streams Query Manager daemon program.
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on the issue:
https://github.com/apache/incubator-rya/pull/272
I'm closing this because https://github.com/apache/incubator-rya/pull/279 takes it over.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167231162
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/InMemoryQueryRepository.java ---
@@ -243,8 +274,68 @@ private void updateCache() {
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");
--- End diff --
Should this log message be moved up into the finally block so it's called if an Exception is thrown?
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167284717
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java ---
@@ -0,0 +1,219 @@
+/**
+ * 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.kafka;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+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.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;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Represents a Kafka Server that manages {@link KafkaQueryChangeLog}s.
+ * <p/>
+ * Thread safe.
--- End diff --
What's the dependency for that annotation?
---
[GitHub] incubator-rya issue #272: RYA-443 Rya Streams Query Manager daemon program.
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on the issue:
https://github.com/apache/incubator-rya/pull/272
How do we want to deal with this project producing an RPM? The ASF jenkins doesn't have the rpm command installed.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167243543
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/kafka/KafkaQueryChangeLogSource.java ---
@@ -0,0 +1,219 @@
+/**
+ * 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.kafka;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+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.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;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Represents a Kafka Server that manages {@link KafkaQueryChangeLog}s.
+ * <p/>
+ * Thread safe.
--- End diff --
Change to <code>@ThreadSafe</code>
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168275659
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java ---
@@ -110,6 +110,16 @@ public boolean equals(final Object o) {
return false;
}
+ @Override
+ public String toString() {
+ return "QueryChange: {" +
+ " Query ID: " + queryId + ",\n" +
+ " Change Type: " + changeType + ",\n" +
+ " Is Active: " + isActive + ",\n" +
+ " SPARQL: " + sparql + "\n" +
--- End diff --
I'm going to assume it is sanitized by the time it reaches this line of code.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167252337
--- Diff: extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java ---
@@ -107,12 +112,11 @@ private String formatQueries(final Set<StreamsQuery> queries) {
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( query.isActive() ? " " : " " )
--- End diff --
Maybe change to:
<code>
.append("Is Active: ")
.append(StringUtils.rightPad(query.isActive(), 9))
</code>
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167234267
--- Diff: extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java ---
@@ -85,11 +92,16 @@ public void initializedWithPopulatedChangeLog() throws Exception {
queries.delete( deletedMeId );
// Create a new totally in memory QueryRepository.
- try(final QueryRepository initializedQueries = new InMemoryQueryRepository( changeLog )) {
+ 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();
--- End diff --
This seems redundant with the outer try-finally block calling stop(). Unless the intent is to test it handling 2 consecutive calls to stop(). The inner try-finally can probably be removed.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167249191
--- Diff: extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/ThreadUtil.java ---
@@ -0,0 +1,43 @@
+/**
+ * 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 {
--- End diff --
Add a private constructor to prevent instantiation
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167284418
--- Diff: extras/rya.streams/query-manager/pom.xml ---
@@ -0,0 +1,253 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.parent</artifactId>
+ <version>3.2.12-incubating-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>rya.streams.query-manager</artifactId>
+
+ <name>Apache Rya Streams Query Manager</name>
+ <description>
+ This module contains the Rya Streams Query Manager.
+ </description>
+
+ <properties>
+ <rpm.staging.path>${project.build.directory}/${project.artifactId}-${project.version}-rpm-staging</rpm.staging.path>
+ </properties>
+
+ <dependencies>
+ <!-- Rya dependencies -->
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.kafka</artifactId>
+ </dependency>
+
+ <!-- Apache Daemon dependencies -->
+ <dependency>
+ <groupId>commons-daemon</groupId>
+ <artifactId>commons-daemon</artifactId>
+ <version>1.1.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.beust</groupId>
+ <artifactId>jcommander</artifactId>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.test.kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <!-- Add the XSD directory as a resource so that it will be packaged in the jar.
+ - Required so that the marshalling code is able to find the schema it is
+ - validating XML against. -->
+ <resources>
+ <resource>
+ <directory>src/main/xsd</directory>
+ </resource>
+ </resources>
+
+ <plugins>
+ <!-- Generate the XML marshalling java code from the XSD file defining its structure. -->
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>jaxb2-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>xjc</id>
+ <goals>
+ <goal>xjc</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <packageName>org.apache.rya.streams.querymanager.xml</packageName>
+ </configuration>
+ </plugin>
+
+ <!-- Ensure the generated java source contains the license header. -->
+ <plugin>
+ <groupId>com.mycila</groupId>
+ <artifactId>license-maven-plugin</artifactId>
+ <configuration>
+ <header>${project.basedir}/src/license/header.txt</header>
+ </configuration>
+ <executions>
+ <execution>
+ <id>update-generated-source-headers</id>
+ <configuration>
+ <basedir>${project.build.directory}/generated-sources/jaxb</basedir>
+ <mapping>
+ <sun-jaxb.episode>XML_STYLE</sun-jaxb.episode>
+ </mapping>
+ </configuration>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>format</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Create a shaded jar that is able to execute the Daemon. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.rya.streams.querymanager.QueryManagerDaemon</mainClass>
+ </transformer>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Use the assembly plugin to create the binary and RPM distributions. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <!-- Stage the files for the RPM construction. -->
+ <execution>
+ <id>stage-content-for-rpms</id>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <!-- This is only staging, so don't attach the built folder to the project. -->
+ <attach>false</attach>
+ <descriptors>
+ <descriptor>src/assembly/rpm-staging.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>rpm-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>set-rpm-properties</id>
+ <goals>
+ <goal>version</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>create-rpm-distribution</id>
--- End diff --
Do you mean the ability to build the software on Windows, or do you mean the ability to install and run Query Manager on windows?
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168277599
--- Diff: extras/rya.streams/query-manager/src/test/java/org/apache/rya/streams/querymanager/ThreadUtil.java ---
@@ -0,0 +1,43 @@
+/**
+ * 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 {
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 closed the pull request at:
https://github.com/apache/incubator-rya/pull/272
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167252837
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java ---
@@ -85,4 +85,19 @@ public boolean equals(final Object o) {
}
return false;
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append("ID: ");
+ sb.append(getQueryId().toString() + "\n");
+ sb.append("Query: ");
+ sb.append(getSparql() + "\n");
--- End diff --
Is "getSparql()" coming from user input? Might want to wrap that in org.apache.rya.api.log.LogUtils.clean(getSparql())
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167241732
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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")
--- End diff --
Wrap query.getSparql() in LogUtils.clean()
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167242868
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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) {
+ 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.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 logWorkQueue - 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 {
--- End diff --
Fix the spacing for this case.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168277066
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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) {
+ 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.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 logWorkQueue - 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 {
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167240341
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 static <T> boolean offerUntilAcceptedOrShutdown(
+ final BlockingQueue<T> workQueue,
+ final T event, final
+ long offerValue,
--- End diff --
final and long are on separate lines
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167232266
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java ---
@@ -110,6 +110,16 @@ public boolean equals(final Object o) {
return false;
}
+ @Override
+ public String toString() {
+ return "QueryChange: {" +
+ " Query ID: " + queryId + ",\n" +
+ " Change Type: " + changeType + ",\n" +
+ " Is Active: " + isActive + ",\n" +
+ " SPARQL: " + sparql + "\n" +
--- End diff --
Is "sparql" coming from user input? Might want to wrap that in org.apache.rya.api.log.LogUtils.clean(sparql)
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167235195
--- Diff: extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java ---
@@ -0,0 +1,90 @@
+/**
+ * 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;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Properties;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TopologyBuilder;
+import org.apache.rya.api.function.projection.RandomUUIDFactory;
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.kafka.topology.TopologyBuilderFactory;
+import org.apache.rya.streams.kafka.topology.TopologyBuilderFactory.TopologyBuilderException;
+import org.apache.rya.streams.kafka.topology.TopologyFactory;
+import org.openrdf.query.MalformedQueryException;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Creates {@link KafkaStreams} objects that are able to process {@link StreamsQuery}s
+ * using a single thread of execution starting from the earliest point in within the
+ * input topic. The Application ID used by the client is based on the Query ID of the
+ * query that is being executed so that this job may resume where it left off if it
+ * is stopped.
+ */
+@DefaultAnnotation(NonNull.class)
+public class SingleThreadKafkaStreamsFactory implements KafkaStreamsFactory {
+
+ private final TopologyBuilderFactory topologyFactory = new TopologyFactory();
+
+ private final String bootstrapServersConfig;
+
+ /**
+ * Constructs an instance of {@link SingleThreadKafkaStreamsFactory}.
+ *
+ * @param bootstrapServersConfig - Configures which Kafka cluster the jobs will interact with. (not null)
+ */
+ public SingleThreadKafkaStreamsFactory(final String bootstrapServersConfig) {
+ this.bootstrapServersConfig = requireNonNull(bootstrapServersConfig);
+ }
+
+ @Override
+ public KafkaStreams make(final String ryaInstance, final StreamsQuery query) throws KafkaStreamsFactoryException {
+ requireNonNull(ryaInstance);
+ requireNonNull(query);
+
+ // Setup the Kafka Stream program.
+ final Properties streamsProps = new Properties();
+
+ // Configure the Kafka servers that will be talked to.
+ streamsProps.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersConfig);
+
+ // Use the Query ID as the Application ID to ensure we resume where we left off the last time this command was run.
+ streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "RyaStreams-Query-" + query.getQueryId());
+
+ // Always start at the beginning of the input topic.
+ streamsProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ // Setup the topology that processes the Query.
+ final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
+ final String resultsTopic = KafkaTopics.queryResultsTopic(query.getQueryId());
+
+ try {
+ final TopologyBuilder topologyBuilder = topologyFactory.build(query.getSparql(), statementsTopic, resultsTopic, new RandomUUIDFactory());
+ return new KafkaStreams(topologyBuilder, new StreamsConfig(streamsProps));
+ } catch (MalformedQueryException | TopologyBuilderException e) {
--- End diff --
add final to Exceptions
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167242070
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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) {
+ 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.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 logWorkQueue - A queue of {@link QueryEvent}s that will be worked by this object. (not null)
--- End diff --
the @param name doesn't match
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276451
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 static <T> boolean offerUntilAcceptedOrShutdown(
+ final BlockingQueue<T> workQueue,
+ final T event, final
+ long offerValue,
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167240617
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
--- End diff --
Fix spacing between last 2 parameters
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r170069054
--- Diff: extras/rya.streams/query-manager/pom.xml ---
@@ -0,0 +1,253 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.parent</artifactId>
+ <version>3.2.12-incubating-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>rya.streams.query-manager</artifactId>
+
+ <name>Apache Rya Streams Query Manager</name>
+ <description>
+ This module contains the Rya Streams Query Manager.
+ </description>
+
+ <properties>
+ <rpm.staging.path>${project.build.directory}/${project.artifactId}-${project.version}-rpm-staging</rpm.staging.path>
+ </properties>
+
+ <dependencies>
+ <!-- Rya dependencies -->
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.kafka</artifactId>
+ </dependency>
+
+ <!-- Apache Daemon dependencies -->
+ <dependency>
+ <groupId>commons-daemon</groupId>
+ <artifactId>commons-daemon</artifactId>
+ <version>1.1.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.beust</groupId>
+ <artifactId>jcommander</artifactId>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.test.kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <!-- Add the XSD directory as a resource so that it will be packaged in the jar.
+ - Required so that the marshalling code is able to find the schema it is
+ - validating XML against. -->
+ <resources>
+ <resource>
+ <directory>src/main/xsd</directory>
+ </resource>
+ </resources>
+
+ <plugins>
+ <!-- Generate the XML marshalling java code from the XSD file defining its structure. -->
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>jaxb2-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>xjc</id>
+ <goals>
+ <goal>xjc</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <packageName>org.apache.rya.streams.querymanager.xml</packageName>
+ </configuration>
+ </plugin>
+
+ <!-- Ensure the generated java source contains the license header. -->
+ <plugin>
+ <groupId>com.mycila</groupId>
+ <artifactId>license-maven-plugin</artifactId>
+ <configuration>
+ <header>${project.basedir}/src/license/header.txt</header>
+ </configuration>
+ <executions>
+ <execution>
+ <id>update-generated-source-headers</id>
+ <configuration>
+ <basedir>${project.build.directory}/generated-sources/jaxb</basedir>
+ <mapping>
+ <sun-jaxb.episode>XML_STYLE</sun-jaxb.episode>
+ </mapping>
+ </configuration>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>format</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Create a shaded jar that is able to execute the Daemon. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.rya.streams.querymanager.QueryManagerDaemon</mainClass>
+ </transformer>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Use the assembly plugin to create the binary and RPM distributions. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <!-- Stage the files for the RPM construction. -->
+ <execution>
+ <id>stage-content-for-rpms</id>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <!-- This is only staging, so don't attach the built folder to the project. -->
+ <attach>false</attach>
+ <descriptors>
+ <descriptor>src/assembly/rpm-staging.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>rpm-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>set-rpm-properties</id>
+ <goals>
+ <goal>version</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>create-rpm-distribution</id>
--- End diff --
Probably both but not needed for this PR.
---
[GitHub] incubator-rya issue #272: RYA-443 Rya Streams Query Manager daemon program.
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit commented on the issue:
https://github.com/apache/incubator-rya/pull/272
Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/687/<h2>Build result: FAILURE</span></h2>[...truncated 3.01 MB...][INFO] Apache Rya Spark Support ........................... SKIPPED[INFO] Apache Rya Web Projects ............................ SKIPPED[INFO] Apache Rya Web Implementation ...................... SKIPPED[INFO] ------------------------------------------------------------------------[INFO] BUILD FAILURE[INFO] ------------------------------------------------------------------------[INFO] Total time: 40:57 min[INFO] Finished at: 2018-02-08T22:25:43+00:00[INFO] Final Memory: 546M/3106M[INFO] ------------------------------------------------------------------------[ERROR] Failed to execute goal org.codehaus.mojo:rpm-maven-plugin:2.1.5:attached-rpm (create-rpm-distribution) on project rya.streams.query-manager: Unable to query for default vendor from RPM: Error while executing process. Cannot run program "rpm": error=2, No such file or directory
-> [Help 1][ERROR] [ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.[ERROR] Re-run Maven using the -X switch to enable full debug logging.[ERROR] [ERROR] For more information about the errors and possible solutions, please read the following articles:[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException[ERROR] [ERROR] After correcting the problems, you can resume the build with the command[ERROR] mvn <goals> -rf :rya.streams.query-managerchannel stoppedSetting status of 50a81217e5b16a4b1377f199aefc3a1c5a225216 to FAILURE with url https://builds.apache.org/job/incubator-rya-master-with-optionals-pull-requests/687/ and message: 'FAILURE 'Using context: Jenkins: clean package -Pgeoindexing
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167283976
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/queries/QueryChange.java ---
@@ -110,6 +110,16 @@ public boolean equals(final Object o) {
return false;
}
+ @Override
+ public String toString() {
+ return "QueryChange: {" +
+ " Query ID: " + queryId + ",\n" +
+ " Change Type: " + changeType + ",\n" +
+ " Is Active: " + isActive + ",\n" +
+ " SPARQL: " + sparql + "\n" +
--- End diff --
What does that function do?
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167230725
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java ---
@@ -85,4 +85,19 @@ public boolean equals(final Object o) {
}
return false;
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append("ID: ");
+ sb.append(getQueryId().toString() + "\n");
+ sb.append("Query: ");
+ sb.append(getSparql() + "\n");
+ sb.append("Is ");
+ if (!isActive) {
+ sb.append(" Not ");
--- End diff --
No space is needed before "Not". The previous append has one.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168278465
--- Diff: extras/rya.streams/client/src/main/java/org/apache/rya/streams/client/command/ListQueriesCommand.java ---
@@ -107,12 +112,11 @@ private String formatQueries(final Set<StreamsQuery> queries) {
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( query.isActive() ? " " : " " )
+ .append("Query: ").append(query.getSparql()).append("\n");
--- End diff --
Should b e sanitized before it gets here.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168278306
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java ---
@@ -85,4 +85,19 @@ public boolean equals(final Object o) {
}
return false;
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append("ID: ");
+ sb.append(getQueryId().toString() + "\n");
+ sb.append("Query: ");
+ sb.append(getSparql() + "\n");
--- End diff --
Nope.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168277168
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManagerDaemon.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 java.util.Objects.requireNonNull;
+
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
+
+import javax.xml.bind.JAXBException;
+
+import org.apache.commons.daemon.Daemon;
+import org.apache.commons.daemon.DaemonContext;
+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;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfig;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfig.PerformanceTunning.QueryChanngeLogDiscoveryPeriod;
+import org.apache.rya.streams.querymanager.xml.QueryManagerConfigUnmarshaller;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xml.sax.SAXException;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.google.common.util.concurrent.AbstractScheduledService.Scheduler;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * JSVC integration code for a {@link QueryManager} to be used as a non-Windows daemon.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManagerDaemon implements Daemon {
+
+ private static final Logger log = LoggerFactory.getLogger(QueryManagerDaemon.class);
+
+ /**
+ * The default configuration file's path for the application.
+ */
+ private static final Path DEFAULT_CONFIGURATION_PATH = Paths.get("config/configuration.xml");
+
+ /**
+ * Command line parameters that are used by all commands that interact with Kafka.
+ */
+ class DaemonParameters {
+ @Parameter(names = {"--config", "-c"}, required = false, description = "The path to the application's configuration file.")
+ public String config;
+ }
+
+ private QueryManager manager = null;
+
+ @Override
+ public void init(final DaemonContext context) throws DaemonInitException, Exception {
+ requireNonNull(context);
+
+ // Parse the command line arguments for the configuration file to use.
+ final String[] args = context.getArguments();
+ final DaemonParameters params = new DaemonParameters();
+ try {
+ new JCommander(params).parse(args);
+ } catch(final ParameterException e) {
+ throw new DaemonInitException("Unable to parse the command line arguments.", e);
+ }
+ final Path configFile = params.config != null ? Paths.get(params.config) : DEFAULT_CONFIGURATION_PATH;
+ log.info("Loading the following configuration file: " + configFile);
+
+ // Unmarshall the configuration file into an object.
+ final QueryManagerConfig config;
+ try(InputStream stream = Files.newInputStream(configFile)) {
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276972
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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) {
+ 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.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 logWorkQueue - A queue of {@link QueryEvent}s that will be worked by this object. (not null)
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168274631
--- Diff: extras/rya.streams/api/src/main/java/org/apache/rya/streams/api/entity/StreamsQuery.java ---
@@ -85,4 +85,19 @@ public boolean equals(final Object o) {
}
return false;
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append("ID: ");
+ sb.append(getQueryId().toString() + "\n");
+ sb.append("Query: ");
+ sb.append(getSparql() + "\n");
+ sb.append("Is ");
+ if (!isActive) {
+ sb.append(" Not ");
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276645
--- Diff: extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java ---
@@ -0,0 +1,884 @@
+/**
+ * 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;
+
+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.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.QueryChangeLog;
+import org.apache.rya.streams.api.queries.QueryChangeLogListener;
+import org.apache.rya.streams.api.queries.QueryRepository;
+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;
+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.
+ * <p>
+ * Only one QueryManager needs to be running to manage any number of rya
+ * instances/rya streams instances.
+ */
+@DefaultAnnotation(NonNull.class)
+public class QueryManager extends AbstractService {
+ private static final Logger log = LoggerFactory.getLogger(QueryManager.class);
+
+ /**
+ * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific
+ * Rya instnace.
+ */
+ private final QueryChangeLogSource changeLogSource;
+
+ /**
+ * The engine that is responsible for executing {@link StreamsQuery}s.
+ */
+ private final QueryExecutor queryExecutor;
+
+ /**
+ * 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 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 long blockingValue,
+ final TimeUnit blockingUnits) {
+ this.changeLogSource = requireNonNull(source);
+ this.queryExecutor = requireNonNull(queryExecutor);
+ Preconditions.checkArgument(blockingValue > 0, "The blocking value must be > 0. Was: " + blockingValue);
+ this.blockingValue = blockingValue;
+ this.blockingUnits = requireNonNull(blockingUnits);
+ }
+
+ @Override
+ 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 {
+ // 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();
+
+ // 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 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 {
+ 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.");
+ }
+
+ /**
+ * 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 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 {
+
+ /**
+ * The types of events that may be observed.
+ */
+ 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 String toString() {
+ return "LogEvent {\n" +
+ " Rya Instance: " + ryaInstance + ",\n" +
+ " Event Type: " + eventType + "\n" +
+ "}";
+ }
+
+ /**
+ * 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));
+ }
+
+ /**
+ * 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());
+ }
+ }
+
+ /**
+ * An observation that a {@link StreamsQuery} needs to be executing or not
+ * via the provided {@link QueryExecutor}.
+ */
+ @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 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")
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168275881
--- Diff: extras/rya.streams/api/src/test/java/org/apache/rya/streams/api/queries/InMemoryQueryRepositoryTest.java ---
@@ -85,11 +92,16 @@ public void initializedWithPopulatedChangeLog() throws Exception {
queries.delete( deletedMeId );
// Create a new totally in memory QueryRepository.
- try(final QueryRepository initializedQueries = new InMemoryQueryRepository( changeLog )) {
+ 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();
--- End diff --
Removed.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167235009
--- Diff: extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/KafkaTopics.java ---
@@ -48,7 +51,28 @@
*/
public static String queryChangeLogTopic(final String ryaInstance) {
requireNonNull(ryaInstance);
- return ryaInstance + "-QueryChangeLog";
+ return ryaInstance + QUERY_CHANGE_LOG_TOPIC_SUFFIX;
+ }
+
+ /**
+ * Get the Rya instance name from a Kafka topic name that has been used for a {@link QueryChnageLog}.
--- End diff --
typo. QueryChangeLog
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by kchilton2 <gi...@git.apache.org>.
Github user kchilton2 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r168276251
--- Diff: extras/rya.streams/kafka/src/main/java/org/apache/rya/streams/kafka/SingleThreadKafkaStreamsFactory.java ---
@@ -0,0 +1,90 @@
+/**
+ * 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;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.Properties;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.TopologyBuilder;
+import org.apache.rya.api.function.projection.RandomUUIDFactory;
+import org.apache.rya.streams.api.entity.StreamsQuery;
+import org.apache.rya.streams.kafka.topology.TopologyBuilderFactory;
+import org.apache.rya.streams.kafka.topology.TopologyBuilderFactory.TopologyBuilderException;
+import org.apache.rya.streams.kafka.topology.TopologyFactory;
+import org.openrdf.query.MalformedQueryException;
+
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * Creates {@link KafkaStreams} objects that are able to process {@link StreamsQuery}s
+ * using a single thread of execution starting from the earliest point in within the
+ * input topic. The Application ID used by the client is based on the Query ID of the
+ * query that is being executed so that this job may resume where it left off if it
+ * is stopped.
+ */
+@DefaultAnnotation(NonNull.class)
+public class SingleThreadKafkaStreamsFactory implements KafkaStreamsFactory {
+
+ private final TopologyBuilderFactory topologyFactory = new TopologyFactory();
+
+ private final String bootstrapServersConfig;
+
+ /**
+ * Constructs an instance of {@link SingleThreadKafkaStreamsFactory}.
+ *
+ * @param bootstrapServersConfig - Configures which Kafka cluster the jobs will interact with. (not null)
+ */
+ public SingleThreadKafkaStreamsFactory(final String bootstrapServersConfig) {
+ this.bootstrapServersConfig = requireNonNull(bootstrapServersConfig);
+ }
+
+ @Override
+ public KafkaStreams make(final String ryaInstance, final StreamsQuery query) throws KafkaStreamsFactoryException {
+ requireNonNull(ryaInstance);
+ requireNonNull(query);
+
+ // Setup the Kafka Stream program.
+ final Properties streamsProps = new Properties();
+
+ // Configure the Kafka servers that will be talked to.
+ streamsProps.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServersConfig);
+
+ // Use the Query ID as the Application ID to ensure we resume where we left off the last time this command was run.
+ streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "RyaStreams-Query-" + query.getQueryId());
+
+ // Always start at the beginning of the input topic.
+ streamsProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ // Setup the topology that processes the Query.
+ final String statementsTopic = KafkaTopics.statementsTopic(ryaInstance);
+ final String resultsTopic = KafkaTopics.queryResultsTopic(query.getQueryId());
+
+ try {
+ final TopologyBuilder topologyBuilder = topologyFactory.build(query.getSparql(), statementsTopic, resultsTopic, new RandomUUIDFactory());
+ return new KafkaStreams(topologyBuilder, new StreamsConfig(streamsProps));
+ } catch (MalformedQueryException | TopologyBuilderException e) {
--- End diff --
Done.
---
[GitHub] incubator-rya pull request #272: RYA-443 Rya Streams Query Manager daemon pr...
Posted by ejwhite922 <gi...@git.apache.org>.
Github user ejwhite922 commented on a diff in the pull request:
https://github.com/apache/incubator-rya/pull/272#discussion_r167236812
--- Diff: extras/rya.streams/query-manager/pom.xml ---
@@ -0,0 +1,253 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.parent</artifactId>
+ <version>3.2.12-incubating-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>rya.streams.query-manager</artifactId>
+
+ <name>Apache Rya Streams Query Manager</name>
+ <description>
+ This module contains the Rya Streams Query Manager.
+ </description>
+
+ <properties>
+ <rpm.staging.path>${project.build.directory}/${project.artifactId}-${project.version}-rpm-staging</rpm.staging.path>
+ </properties>
+
+ <dependencies>
+ <!-- Rya dependencies -->
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.streams.kafka</artifactId>
+ </dependency>
+
+ <!-- Apache Daemon dependencies -->
+ <dependency>
+ <groupId>commons-daemon</groupId>
+ <artifactId>commons-daemon</artifactId>
+ <version>1.1.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.beust</groupId>
+ <artifactId>jcommander</artifactId>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.test.kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <!-- Add the XSD directory as a resource so that it will be packaged in the jar.
+ - Required so that the marshalling code is able to find the schema it is
+ - validating XML against. -->
+ <resources>
+ <resource>
+ <directory>src/main/xsd</directory>
+ </resource>
+ </resources>
+
+ <plugins>
+ <!-- Generate the XML marshalling java code from the XSD file defining its structure. -->
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>jaxb2-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>xjc</id>
+ <goals>
+ <goal>xjc</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <packageName>org.apache.rya.streams.querymanager.xml</packageName>
+ </configuration>
+ </plugin>
+
+ <!-- Ensure the generated java source contains the license header. -->
+ <plugin>
+ <groupId>com.mycila</groupId>
+ <artifactId>license-maven-plugin</artifactId>
+ <configuration>
+ <header>${project.basedir}/src/license/header.txt</header>
+ </configuration>
+ <executions>
+ <execution>
+ <id>update-generated-source-headers</id>
+ <configuration>
+ <basedir>${project.build.directory}/generated-sources/jaxb</basedir>
+ <mapping>
+ <sun-jaxb.episode>XML_STYLE</sun-jaxb.episode>
+ </mapping>
+ </configuration>
+ <phase>process-sources</phase>
+ <goals>
+ <goal>format</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Create a shaded jar that is able to execute the Daemon. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.rya.streams.querymanager.QueryManagerDaemon</mainClass>
+ </transformer>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Use the assembly plugin to create the binary and RPM distributions. -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <!-- Stage the files for the RPM construction. -->
+ <execution>
+ <id>stage-content-for-rpms</id>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <phase>package</phase>
+ <configuration>
+ <!-- This is only staging, so don't attach the built folder to the project. -->
+ <attach>false</attach>
+ <descriptors>
+ <descriptor>src/assembly/rpm-staging.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>rpm-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>set-rpm-properties</id>
+ <goals>
+ <goal>version</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>create-rpm-distribution</id>
--- End diff --
We'll eventually need Windows support. But this is fine for now.
---