{"action":"created","comment":{"url":"https://api.github.com/repos/apache/incubator-rya/pulls/comments/168276972","pull_request_review_id":96618371,"id":168276972,"diff_hunk":"@@ -0,0 +1,884 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *   http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing,\n+ * software distributed under the License is distributed on an\n+ * \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n+ * KIND, either express or implied.  See the License for the\n+ * specific language governing permissions and limitations\n+ * under the License.\n+ */\n+package org.apache.rya.streams.querymanager;\n+\n+import static java.util.Objects.requireNonNull;\n+\n+import java.util.HashMap;\n+import java.util.Map;\n+import java.util.Objects;\n+import java.util.Optional;\n+import java.util.Set;\n+import java.util.UUID;\n+import java.util.concurrent.ArrayBlockingQueue;\n+import java.util.concurrent.BlockingQueue;\n+import java.util.concurrent.CountDownLatch;\n+import java.util.concurrent.ExecutorService;\n+import java.util.concurrent.Executors;\n+import java.util.concurrent.RejectedExecutionException;\n+import java.util.concurrent.TimeUnit;\n+import java.util.concurrent.atomic.AtomicBoolean;\n+\n+import org.apache.rya.streams.api.entity.StreamsQuery;\n+import org.apache.rya.streams.api.queries.ChangeLogEntry;\n+import org.apache.rya.streams.api.queries.InMemoryQueryRepository;\n+import org.apache.rya.streams.api.queries.QueryChange;\n+import org.apache.rya.streams.api.queries.QueryChangeLog;\n+import org.apache.rya.streams.api.queries.QueryChangeLogListener;\n+import org.apache.rya.streams.api.queries.QueryRepository;\n+import org.apache.rya.streams.querymanager.QueryChangeLogSource.SourceListener;\n+import org.apache.rya.streams.querymanager.QueryExecutor.QueryExecutorException;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import com.google.common.base.Preconditions;\n+import com.google.common.util.concurrent.AbstractScheduledService.Scheduler;\n+import com.google.common.util.concurrent.AbstractService;\n+import com.google.common.util.concurrent.UncheckedExecutionException;\n+\n+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;\n+import edu.umd.cs.findbugs.annotations.NonNull;\n+\n+/**\n+ * A service for managing {@link StreamsQuery} running on a Rya Streams system.\n+ * <p>\n+ * Only one QueryManager needs to be running to manage any number of rya\n+ * instances/rya streams instances.\n+ */\n+@DefaultAnnotation(NonNull.class)\n+public class QueryManager extends AbstractService {\n+    private static final Logger log = LoggerFactory.getLogger(QueryManager.class);\n+\n+    /**\n+     * The source of {@link QueryChangeLog}s. Each log discovered is bound to a specific\n+     * Rya instnace.\n+     */\n+    private final QueryChangeLogSource changeLogSource;\n+\n+    /**\n+     * The engine that is responsible for executing {@link StreamsQuery}s.\n+     */\n+    private final QueryExecutor queryExecutor;\n+\n+    /**\n+     * How long blocking operations will be attempted before potentially trying again.\n+     */\n+    private final long blockingValue;\n+\n+    /**\n+     * The units for {@link #blockingValue}.\n+     */\n+    private final TimeUnit blockingUnits;\n+\n+    /**\n+     * Used to inform threads that the application is shutting down, so they must stop work.\n+     */\n+    private final AtomicBoolean shutdownSignal = new AtomicBoolean(false);\n+\n+    /**\n+     * This thread pool manages the two thread used to work the {@link LogEvent}s\n+     * and the {@link QueryEvent}s.\n+     */\n+    private final ExecutorService executor = Executors.newFixedThreadPool(2);\n+\n+    /**\n+     * Creates a new {@link QueryManager}.\n+     *\n+     * @param queryExecutor - Runs the active {@link StreamsQuery}s. (not null)\n+     * @param source - The {@link QueryChangeLogSource} of QueryChangeLogs. (not null)\n+     * @param blockingValue - How long blocking operations will try before looping. (> 0)\n+     * @param blockingUnits - The units of the {@code blockingValue}. (not null)\n+     */\n+    public QueryManager(\n+            final QueryExecutor queryExecutor,\n+            final QueryChangeLogSource source,\n+            final long blockingValue,\n+            final TimeUnit blockingUnits) {\n+        this.changeLogSource = requireNonNull(source);\n+        this.queryExecutor = requireNonNull(queryExecutor);\n+        Preconditions.checkArgument(blockingValue > 0, \"The blocking value must be > 0. Was: \" + blockingValue);\n+        this.blockingValue = blockingValue;\n+        this.blockingUnits = requireNonNull(blockingUnits);\n+    }\n+\n+    @Override\n+    protected void doStart() {\n+        log.info(\"Starting a QueryManager.\");\n+\n+        // A work queue of discovered Query Change Logs that need to be handled.\n+        // This queue exists so that the source notifying thread may be released\n+        // immediately instead of calling into blocking functions.\n+        final BlockingQueue<LogEvent> logEvents = new ArrayBlockingQueue<>(1024);\n+\n+        // A work queue of discovered Query Changes from the monitored Query Change Logs\n+        // that need to be handled. This queue exists so that the Query Repository notifying\n+        // thread may be released immediately instead of calling into blocking functions.\n+        final BlockingQueue<QueryEvent> queryEvents = new ArrayBlockingQueue<>(1024);\n+\n+        try {\n+            // Start up a LogEventWorker using the executor service.\n+            executor.submit(new LogEventWorker(logEvents, queryEvents, blockingValue, blockingUnits, shutdownSignal));\n+\n+            // Start up a QueryEvent Worker using the executor service.\n+            executor.submit(new QueryEventWorker(queryEvents, queryExecutor, blockingValue, blockingUnits, shutdownSignal));\n+\n+            // Start up the query execution framework.\n+            queryExecutor.startAndWait();\n+\n+            // Startup the source that discovers new Query Change Logs.\n+            changeLogSource.startAndWait();\n+\n+            // Subscribe the source a listener that writes to the LogEventWorker's work queue.\n+            changeLogSource.subscribe(new LogEventWorkGenerator(logEvents, blockingValue, blockingUnits, shutdownSignal));\n+        } catch(final RejectedExecutionException | UncheckedExecutionException e) {\n+            log.error(\"Could not start up a QueryManager.\", e);\n+            notifyFailed(e);\n+        }\n+\n+        // Notify the service was successfully started.\n+        notifyStarted();\n+\n+        log.info(\"QueryManager has finished starting.\");\n+    }\n+\n+    @Override\n+    protected void doStop() {\n+        log.info(\"Stopping a QueryManager.\");\n+\n+        // Set the shutdown flag so that all components that rely on that signal will stop processing.\n+        shutdownSignal.set(true);\n+\n+        // Stop the workers and wait for them to die.\n+        executor.shutdownNow();\n+        try {\n+            if(!executor.awaitTermination(10, TimeUnit.SECONDS)) {\n+                log.warn(\"Waited 10 seconds for the worker threads to die, but they are still running.\");\n+            }\n+        } catch (final InterruptedException e) {\n+            log.warn(\"Waited 10 seconds for the worker threads to die, but they are still running.\");\n+        }\n+\n+        // Stop the source of new Change Logs.\n+        try {\n+            changeLogSource.stopAndWait();\n+        } catch(final UncheckedExecutionException e) {\n+            log.warn(\"Could not stop the Change Log Source.\", e);\n+        }\n+\n+        // Stop the query execution framework.\n+        try {\n+            queryExecutor.stopAndWait();\n+        } catch(final UncheckedExecutionException e) {\n+            log.warn(\"Could not stop the Query Executor\", e);\n+        }\n+\n+        // Notify the service was successfully stopped.\n+        notifyStopped();\n+\n+        log.info(\"QueryManager has finished stopping.\");\n+    }\n+\n+    /**\n+     * Offer a unit of work to a blocking queue until it is either accepted, or the\n+     * shutdown signal is set.\n+     *\n+     * @param workQueue - The blocking work queue to write to. (not null)\n+     * @param event - The event that will be offered to the work queue. (not null)\n+     * @param offerValue - How long to wait when offering new work.\n+     * @param offerUnits - The unit for the {@code offerValue}. (not null)\n+     * @param shutdownSignal - Used to signal application shutdown has started, so\n+     *   this method may terminate without ever placing the event on the queue. (not null)\n+     * @return {@code true} if the evet nwas added to the queue, otherwise false.\n+     */\n+    private static <T> boolean offerUntilAcceptedOrShutdown(\n+            final BlockingQueue<T> workQueue,\n+            final T event, final\n+            long offerValue,\n+            final TimeUnit offerUnits,\n+            final AtomicBoolean shutdownSignal) {\n+        requireNonNull(workQueue);\n+        requireNonNull(event);\n+        requireNonNull(shutdownSignal);\n+\n+        boolean submitted = false;\n+        while(!submitted && !shutdownSignal.get()) {\n+            try {\n+                submitted = workQueue.offer(event, offerValue, offerUnits);\n+                if(!submitted) {\n+                    log.debug(\"An event could not be added to a work queue after waiting 5 seconds. Trying again...\");\n+                }\n+            } catch (final InterruptedException e) {\n+                log.debug(\"An event could not be added to a work queue after waiting 5 seconds. Trying again...\");\n+            }\n+        }\n+        return submitted;\n+    }\n+\n+    /**\n+     * An observation that a {@link QueryChangeLog} was created within or\n+     * removed from a {@link QueryChangeLogSource}.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class LogEvent {\n+\n+        /**\n+         * The types of events that may be observed.\n+         */\n+        static enum LogEventType {\n+            /**\n+             * A {@link QueryChangeLog} was created within a {@link QueryChangeLogSource}.\n+             */\n+            CREATE,\n+\n+            /**\n+             * A {@link QueryChangeLog} was deleted from a {@link QueryChangeLogSource}.\n+             */\n+            DELETE;\n+        }\n+\n+        private final String ryaInstance;\n+        private final LogEventType eventType;\n+        private final Optional<QueryChangeLog> log;\n+\n+        /**\n+         * Constructs an instance of {@link LogEvent}.\n+         *\n+         * @param ryaInstance - The Rya Instance the log is/was for. (not null)\n+         * @param eventType - The type of event that was observed. (not null)\n+         * @param log - The log if this is a create event. (not null)\n+         */\n+        private LogEvent(final String ryaInstance, final LogEventType eventType, final Optional<QueryChangeLog> log) {\n+            this.ryaInstance = requireNonNull(ryaInstance);\n+            this.eventType = requireNonNull(eventType);\n+            this.log = requireNonNull(log);\n+        }\n+\n+        /**\n+         * @return The Rya Instance whose log was either created or deleted.\n+         */\n+        public String getRyaInstanceName() {\n+            return ryaInstance;\n+        }\n+\n+        /**\n+         * @return The type of event that was observed.\n+         */\n+        public LogEventType getEventType() {\n+            return eventType;\n+        }\n+\n+        /**\n+         * @return The {@link QueryChangeLog} if this is a CREATE event.\n+         */\n+        public Optional<QueryChangeLog> getQueryChangeLog() {\n+            return log;\n+        }\n+\n+        @Override\n+        public String toString() {\n+            return \"LogEvent {\\n\" +\n+                   \"    Rya Instance: \" + ryaInstance + \",\\n\" +\n+                   \"    Event Type: \" + eventType + \"\\n\" +\n+                   \"}\";\n+        }\n+\n+        /**\n+         * Make a {@link LogEvent} that indicates a {@link QueryChangeLog} was created within a\n+         * {@link QueryChangeLogSource}.\n+         *\n+         * @param ryaInstance - The Rya Instance the created log is for. (not null)\n+         * @param log - The created {@link QueryChangeLog. (not null)\n+         * @return A {@link LogEvent} built using the provided values.\n+         */\n+        public static LogEvent create(final String ryaInstance, final QueryChangeLog log) {\n+            return new LogEvent(ryaInstance, LogEventType.CREATE ,Optional.of(log));\n+        }\n+\n+        /**\n+         * Make a {@link LogEvent} that indicates a {@link QueryChangeLog} was deleted from\n+         * a {@link QueryChangeLogSource}.\n+         *\n+         * @param ryaInstance - The Rya Instance whose log was deleted. (not null)\n+         * @return A {@link LogEvent} built using the provided values.\n+         */\n+        public static LogEvent delete(final String ryaInstance) {\n+            return new LogEvent(ryaInstance, LogEventType.DELETE, Optional.empty());\n+        }\n+    }\n+\n+    /**\n+     * An observation that a {@link StreamsQuery} needs to be executing or not\n+     * via the provided {@link QueryExecutor}.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class QueryEvent {\n+\n+        /**\n+         * The type of events that may be observed.\n+         */\n+        public static enum QueryEventType {\n+            /**\n+             * Indicates a {@link StreamsQuery} needs to be executing.\n+             */\n+            EXECUTING,\n+\n+            /**\n+             * Indicates a {@link StreamsQuery} needs to be stopped.\n+             */\n+            STOPPED,\n+\n+            /**\n+             * Indicates all {@link StreamsQuery}s for a Rya instance need to be stopped.\n+             */\n+            STOP_ALL;\n+        }\n+\n+        private final String ryaInstance;\n+        private final QueryEventType type;\n+        private final Optional<UUID> queryId;\n+        private final Optional<StreamsQuery> query;\n+\n+        /**\n+         * Constructs an instance of {@link QueryEvent}.\n+         *\n+         * @param ryaInstance - The Rya instance that generated the event. (not null)\n+         * @param type - Indicates whether the query needs to be executing or not. (not null)\n+         * @param queryId - If stopped, the ID of the query that must not be running. (not null)\n+         * @param query - If executing, the StreamsQuery that defines what should be executing. (not null)\n+         */\n+        private QueryEvent(\n+                final String ryaInstance,\n+                final QueryEventType type,\n+                final Optional<UUID> queryId,\n+                final Optional<StreamsQuery> query) {\n+            this.ryaInstance = requireNonNull(ryaInstance);\n+            this.type = requireNonNull(type);\n+            this.queryId = requireNonNull(queryId);\n+            this.query = requireNonNull(query);\n+        }\n+\n+        /**\n+         * @return The Rya instance that generated the event.\n+         */\n+        public String getRyaInstance() {\n+            return ryaInstance;\n+        }\n+\n+        /**\n+         * @return Indicates whether the query needs to be executing or not.\n+         */\n+        public QueryEventType getType() {\n+            return type;\n+        }\n+\n+        /**\n+         * @return If stopped, the ID of the query that must not be running. Otherwise absent.\n+         */\n+        public Optional<UUID> getQueryId() {\n+            return queryId;\n+        }\n+\n+        /**\n+         * @return If executing, the StreamsQuery that defines what should be executing. Otherwise absent.\n+         */\n+        public Optional<StreamsQuery> getStreamsQuery() {\n+            return query;\n+        }\n+\n+        @Override\n+        public int hashCode() {\n+            return Objects.hash(ryaInstance, type, queryId, query);\n+        }\n+\n+        @Override\n+        public boolean equals(final Object o) {\n+            if(o instanceof QueryEvent) {\n+                final QueryEvent other = (QueryEvent) o;\n+                return Objects.equals(ryaInstance, other.ryaInstance) &&\n+                        Objects.equals(type, other.type) &&\n+                        Objects.equals(queryId, other.queryId) &&\n+                        Objects.equals(query, other.query);\n+            }\n+            return false;\n+        }\n+\n+        @Override\n+        public String toString() {\n+            final StringBuilder string = new StringBuilder();\n+            string.append(\"Query Event {\\n\")\n+                  .append(\"    Rya Instance:\").append(ryaInstance).append(\",\\n\")\n+                  .append(\"    Type: \").append(type).append(\",\\n\");\n+            switch(type) {\n+                case EXECUTING:\n+                    append(string, query.get());\n+                    break;\n+                case STOPPED:\n+                    string.append(\"    Query ID: \").append(queryId.get()).append(\"\\n\");\n+                    break;\n+                case STOP_ALL:\n+                    break;\n+                default:\n+                    // Default to showing everything that is in the object.\n+                    string.append(\"    Query ID: \").append(queryId.get()).append(\"\\n\");\n+                    append(string, query.get());\n+                    break;\n+            }\n+            string.append(\"}\");\n+            return string.toString();\n+        }\n+\n+        private void append(final StringBuilder string, final StreamsQuery query) {\n+            requireNonNull(string);\n+            requireNonNull(query);\n+            string.append(\"    Streams Query {\\n\")\n+                  .append(\"        Query ID: \").append(query.getQueryId()).append(\",\\n\")\n+                  .append(\"        Is Active: \").append(query.isActive()).append(\",\\n\")\n+                  .append(\"        SPARQL: \").append(query.getSparql()).append(\"\\n\")\n+                  .append(\"    }\");\n+        }\n+\n+        /**\n+         * Create a {@link QueryEvent} that indicates a query needs to be executing.\n+         *\n+         * @param ryaInstance - The Rya instance that generated the event. (not null)\n+         * @param query - The StreamsQuery that defines what should be executing. (not null)\n+         * @return A {@link QueryEvent} built using the provided values.\n+         */\n+        public static QueryEvent executing(final String ryaInstance, final StreamsQuery query) {\n+            return new QueryEvent(ryaInstance, QueryEventType.EXECUTING, Optional.empty(), Optional.of(query));\n+        }\n+\n+        /**\n+         * Create a {@link QueryEvent} that indicates a query needs to be stopped.\n+         *\n+         * @param ryaInstance - The Rya instance that generated the event. (not null)\n+         * @param queryId - The ID of the query that must not be running. (not null)\n+         * @return A {@link QueryEvent} built using the provided values.\n+         */\n+        public static QueryEvent stopped(final String ryaInstance, final UUID queryId) {\n+            return new QueryEvent(ryaInstance, QueryEventType.STOPPED, Optional.of(queryId), Optional.empty());\n+        }\n+\n+        /**\n+         * Create a {@link QueryEvent} that indicates all queries for a Rya instance needs to be stopped.\n+         *\n+         * @param ryaInstance - The Rya instance that generated the event. (not null)\n+         * @return A {@link QueryEvent} built using the provided values.\n+         */\n+        public static QueryEvent stopALL(final String ryaInstance) {\n+            return new QueryEvent(ryaInstance, QueryEventType.STOP_ALL, Optional.empty(), Optional.empty());\n+        }\n+    }\n+\n+    /**\n+     * Listens to a {@link QueryChangeLogSource} and adds observations to the provided\n+     * work queue. It does so until the provided shutdown signal is set.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class LogEventWorkGenerator implements SourceListener {\n+\n+        private final BlockingQueue<LogEvent> workQueue;\n+        private final AtomicBoolean shutdownSignal;\n+        private final long offerValue;\n+        private final TimeUnit offerUnits;\n+\n+        /**\n+         * Constructs an instance of {@link QueryManagerSourceListener}.\n+         *\n+         * @param workQueue - A blocking queue that will have {@link LogEvent}s offered to it. (not null)\n+         * @param offerValue - How long to wait when offering new work.\n+         * @param offerUnits - The unit for the {@code offerValue}. (not null)\n+         * @param shutdownSignal - Indicates to this listener that it needs to stop adding events\n+         *   to the work queue because the application is shutting down. (not null)\n+         */\n+        public LogEventWorkGenerator(\n+                final BlockingQueue<LogEvent> workQueue,\n+                final long offerValue,\n+                final TimeUnit offerUnits,\n+                final AtomicBoolean shutdownSignal) {\n+            this.workQueue = requireNonNull(workQueue);\n+            this.shutdownSignal = requireNonNull(shutdownSignal);\n+            this.offerValue = offerValue;\n+            this.offerUnits = requireNonNull(offerUnits);\n+        }\n+\n+        @Override\n+        public void notifyCreate(final String ryaInstanceName, final QueryChangeLog changeLog) {\n+            log.info(\"A new Query Change Log has been discovered for Rya Instance \" + ryaInstanceName + \". All \" +\n+                    \"queries that are set to active within it will be started.\");\n+\n+            // Create an event that summarizes this notification.\n+            final LogEvent event = LogEvent.create(ryaInstanceName, changeLog);\n+\n+            // Offer it to the worker until there is room for it in the work queue, or we are shutting down.\n+            offerUntilAcceptedOrShutdown(workQueue, event, offerValue, offerUnits, shutdownSignal);\n+        }\n+\n+        @Override\n+        public void notifyDelete(final String ryaInstanceName) {\n+            log.info(\"The Query Change Log for Rya Instance \" + ryaInstanceName + \" has been deleted. All of the \" +\n+                    \"queries related to that instance will be stopped.\");\n+\n+            // Create an event that summarizes this notification.\n+            final LogEvent event = LogEvent.delete(ryaInstanceName);\n+\n+            // Offer it to the worker until there is room for it in the work queue, or we are shutting down.\n+            offerUntilAcceptedOrShutdown(workQueue, event, offerValue, offerUnits, shutdownSignal);\n+        }\n+    }\n+\n+    /**\n+     * Processes a work queue of {@link LogEvent}s.\n+     * <p/>\n+     * Whenever a new log has been created, then it registers a {@link QueryEventWorkGenerator}\n+     * that generates {@link QueryEvent}s based on the content and updates to the discovered\n+     * {@link QueryChagneLog}.\n+     * <p/>\n+     * Whenever a log is deleted, then the generator is stopped and a stop all {@link QueryEvent}\n+     * is written to the work queue.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class LogEventWorker implements Runnable {\n+\n+        /**\n+         * A map of Rya Instance name to he Query Repository for that instance.\n+         */\n+        private final Map<String, QueryRepository> repos = new HashMap<>();\n+\n+        private final BlockingQueue<LogEvent> logWorkQueue;\n+        private final BlockingQueue<QueryEvent> queryWorkQueue;\n+        private final long blockingValue;\n+        private final TimeUnit blockingUnits;\n+        private final AtomicBoolean shutdownSignal;\n+\n+        /**\n+         * Constructs an instance of {@link LogEventWorker}.\n+         *\n+         * @param logWorkQueue - A queue of {@link LogEvent}s that will be worked by this object. (not null)\n+         * @param queryWorkQueue - A queue where {@link QueryEvent}s will be placed by this object. (not null)\n+         * @param blockingValue - How long to wait when polling/offering new work.\n+         * @param blockingUnits - The unit for the {@code blockingValue}. (not null)\n+         * @param shutdownSignal - Indicates when the application has been shutdown, so the executing thread\n+         *   may exit the {@link #run()} method. (not null)\n+         */\n+        public LogEventWorker(\n+                final BlockingQueue<LogEvent> logWorkQueue,\n+                final BlockingQueue<QueryEvent> queryWorkQueue,\n+                final long blockingValue,\n+                final TimeUnit blockingUnits,\n+                final AtomicBoolean shutdownSignal) {\n+            this.logWorkQueue = requireNonNull(logWorkQueue);\n+            this.queryWorkQueue = requireNonNull(queryWorkQueue);\n+            this.blockingValue = blockingValue;\n+            this.blockingUnits = requireNonNull(blockingUnits);\n+            this.shutdownSignal = requireNonNull(shutdownSignal);\n+        }\n+\n+        @Override\n+        public void run() {\n+            // Run until the shutdown signal is set.\n+            while(!shutdownSignal.get()) {\n+                try {\n+                    // Pull a unit of work from the queue.\n+                    log.debug(\"LogEventWorker - Polling the work queue for a new LogEvent.\");\n+                    final LogEvent logEvent = logWorkQueue.poll(blockingValue, blockingUnits);\n+                    if(logEvent == null) {\n+                        // Poll again if nothing was found.\n+                        continue;\n+                    }\n+\n+                    log.info(\"LogEventWorker - handling: \\n\" + logEvent);\n+                    final String ryaInstance = logEvent.getRyaInstanceName();\n+\n+                    switch(logEvent.getEventType()) {\n+                        case CREATE:\n+                            // If we see a create message for a Rya Instance we are already maintaining,\n+                            // then don't do anything.\n+                            if(repos.containsKey(ryaInstance)) {\n+                                log.warn(\"LogEventWorker - A repository is already being managed for the Rya Instance \" +\n+                                        ryaInstance + \". This message will be ignored.\");\n+                                continue;\n+                            }\n+\n+                            // Create and start a QueryRepository for the discovered log. Hold onto the repository\n+                            // so that it may be shutdown later.\n+                            final Scheduler scheduler = Scheduler.newFixedRateSchedule(0, blockingValue, blockingUnits);\n+                            final QueryRepository repo = new InMemoryQueryRepository(logEvent.getQueryChangeLog().get(), scheduler);\n+                            repo.startAndWait();\n+                            repos.put(ryaInstance, repo);\n+\n+                            // Subscribe a worker that adds the Query Events to the queryWorkQueue queue.\n+                            // A count down latch is used to ensure the returned set of queries are handled\n+                            // prior to any notifications from the repository.\n+                            final CountDownLatch subscriptionWorkFinished = new CountDownLatch(1);\n+                            final QueryEventWorkGenerator queryWorkGenerator =\n+                                    new QueryEventWorkGenerator(ryaInstance, subscriptionWorkFinished, queryWorkQueue,\n+                                            blockingValue, blockingUnits, shutdownSignal);\n+\n+                            log.debug(\"LogEventWorker - Setting up a QueryWorkGenerator...\");\n+                            final Set<StreamsQuery> queries = repo.subscribe(queryWorkGenerator);\n+                            log.debug(\"LogEventWorker - Finished setting up a QueryWorkGenerator.\");\n+\n+                            // Handle the view of the queries within the repository as it existed when\n+                            // the subscription was registered.\n+                            queries.stream()\n+                            .forEach(query -> {\n+                                // Create a QueryEvent that represents the active state of the existing query.\n+                                final QueryEvent queryEvent = query.isActive() ?\n+                                        QueryEvent.executing(ryaInstance, query) : QueryEvent.stopped(ryaInstance, query.getQueryId());\n+                                log.debug(\"LogEventWorker - offering: \" + queryEvent);\n+\n+                                // Offer it to the worker until there is room for it in the work queue, or we are shutting down.\n+                                offerUntilAcceptedOrShutdown(queryWorkQueue, queryEvent, blockingValue, blockingUnits, shutdownSignal);\n+                            });\n+\n+                            // Indicate the subscription work is finished so that the registered listener may start\n+                            // adding work to the queue.\n+                            log.info(\"LogEventWorker - Counting down the subscription work latch.\");\n+                            subscriptionWorkFinished.countDown();\n+                            break;\n+\n+                        case DELETE:\n+                            if(repos.containsKey(ryaInstance)) {\n+                                // Shut down the query repository for the Rya instance. This ensures the listener will\n+                                // not receive any more work that needs to be done.\n+                                final QueryRepository deletedRepo = repos.remove(ryaInstance);\n+                                deletedRepo.stopAndWait();\n+\n+                                // Add work that stops all of the queries related to the instance.\n+                                final QueryEvent stopAllEvent = QueryEvent.stopALL(ryaInstance);\n+                                offerUntilAcceptedOrShutdown(queryWorkQueue, stopAllEvent, blockingValue, blockingUnits, shutdownSignal);\n+                            }\n+                            break;\n+                    }\n+                } catch (final InterruptedException e) {\n+                    log.debug(\"LogEventWorker did not see any new events over the past 5 seconds. Polling again...\");\n+                }\n+            }\n+\n+            log.info(\"LogEventWorker shutting down...\");\n+\n+            // Shutdown all of the QueryRepositories that were started.\n+            repos.values().forEach(repo -> repo.stopAndWait());\n+\n+            log.info(\"LogEventWorker shut down.\");\n+        }\n+    }\n+\n+    /**\n+     * Listens to a {@link QueryRepository} and adds observations to the provided work queue.\n+     * It does so until the provided shutdown signal is set.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class QueryEventWorkGenerator implements QueryChangeLogListener {\n+\n+        private final String ryaInstance;\n+        private final CountDownLatch subscriptionWorkFinished;\n+        private final BlockingQueue<QueryEvent> queryWorkQueue;\n+        private final long blockingValue;\n+        private final TimeUnit blockingUnits;\n+        private final AtomicBoolean shutdownSignal;\n+\n+        /**\n+         * Constructs an instance of {@link QueryEventWorkGenerator}.\n+         *\n+         * @param ryaInstance - The rya instance whose log this objects is watching. (not null)\n+         * @param subscriptionWorkFinished - Indicates when work that needs to be completed before this\n+         *   listener handles notifications is completed. (not null)\n+         * @param queryWorkQueue - A queue where {@link QueryEvent}s will be placed by this object. (not null)\n+         * @param blockingValue - How long to wait when polling/offering new work.\n+         * @param blockingUnits - The unit for the {@code blockingValue}. (not null)\n+         * @param shutdownSignal - Indicates to this listener that it needs to stop adding events\n+         *   to the work queue because the application is shutting down. (not null)\n+         */\n+        public QueryEventWorkGenerator(\n+                final String ryaInstance,\n+                final CountDownLatch subscriptionWorkFinished,\n+                final BlockingQueue<QueryEvent> queryWorkQueue,\n+                final long blockingValue,\n+                final TimeUnit blockingUnits,\n+                final AtomicBoolean shutdownSignal) {\n+            this.ryaInstance = requireNonNull(ryaInstance);\n+            this.subscriptionWorkFinished = requireNonNull(subscriptionWorkFinished);\n+            this.queryWorkQueue = requireNonNull(queryWorkQueue);\n+            this.blockingValue = blockingValue;\n+            this.blockingUnits = requireNonNull(blockingUnits);\n+            this.shutdownSignal = requireNonNull(shutdownSignal);\n+        }\n+\n+        @Override\n+        public void notify(final ChangeLogEntry<QueryChange> queryChangeEvent, final Optional<StreamsQuery> newQueryState) {\n+            requireNonNull(queryChangeEvent);\n+            requireNonNull(newQueryState);\n+\n+            // Wait for the subscription work to be finished.\n+            try {\n+                log.debug(\"Waiting for Subscription Work Finished latch to release...\");\n+                while(!shutdownSignal.get() && !subscriptionWorkFinished.await(blockingValue, blockingUnits)) {\n+                    log.debug(\"Still waiting...\");\n+                }\n+                log.debug(\"Subscription Work Finished latch to released.\");\n+            } catch (final InterruptedException e) {\n+                log.warn(\"Interrupted while waiting for the Subscription Work Finished latch to be \" +\n+                        \"released. Shutting down?\", e);\n+            }\n+\n+            // If we left the loop because of a shutdown, return immediately.\n+            if(shutdownSignal.get()) {\n+                log.debug(\"Not processing notification. Shutting down.\");\n+                return;\n+            }\n+\n+            // Generate work from the notification.\n+            final QueryChange change = queryChangeEvent.getEntry();\n+            switch(change.getChangeType()) {\n+                case CREATE:\n+                    if(newQueryState.isPresent()) {\n+                        log.info(\"Rya Instance \" + ryaInstance + \" created Rya Streams query \" + newQueryState + \".\");\n+                        final StreamsQuery newQuery = newQueryState.get();\n+                        if(newQuery.isActive()) {\n+                            final QueryEvent executeNewQuery = QueryEvent.executing(ryaInstance, newQuery);\n+                            offerUntilAcceptedOrShutdown(queryWorkQueue, executeNewQuery, blockingValue, blockingUnits, shutdownSignal);\n+                        }\n+                    } else {\n+                        log.error(\"Received a CREATE QueryChange for Rya Instance: \" + ryaInstance +\n+                                \", Query ID: \" + change.getQueryId() + \", but the QueryRepository did not supply a \" +\n+                                \"StreamsQuery representing the created query. The query will not be processed.\");\n+                    }\n+                    break;\n+\n+                case DELETE:\n+                    final UUID deletedQueryId = change.getQueryId();\n+                    log.info(\"Rya Instance \" + ryaInstance + \" deleted Rya Streams query with ID \" + deletedQueryId);\n+                    final QueryEvent stopDeletedQuery = QueryEvent.stopped(ryaInstance, deletedQueryId);\n+                    offerUntilAcceptedOrShutdown(queryWorkQueue, stopDeletedQuery, blockingValue, blockingUnits, shutdownSignal);\n+                    break;\n+\n+                case UPDATE:\n+                    if(newQueryState.isPresent()) {\n+                        final StreamsQuery updatedQuery = newQueryState.get();\n+                        if(updatedQuery.isActive()) {\n+                            log.info(\"Rya Instance \" + ryaInstance + \" updated Rya Streams query with ID \" +\n+                                    updatedQuery.getQueryId() + \" to be active.\");\n+                            final QueryEvent executeUpdatedQuery = QueryEvent.executing(ryaInstance, updatedQuery);\n+                            offerUntilAcceptedOrShutdown(queryWorkQueue, executeUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);\n+                        } else {\n+                            log.info(\"Rya Instance \" + ryaInstance + \" updated Rya Streams query with ID \" +\n+                                    updatedQuery.getQueryId() + \" to be inactive.\");\n+                            final QueryEvent stopUpdatedQuery = QueryEvent.stopped(ryaInstance, updatedQuery.getQueryId());\n+                            offerUntilAcceptedOrShutdown(queryWorkQueue, stopUpdatedQuery, blockingValue, blockingUnits, shutdownSignal);\n+                        }\n+                    } else {\n+                        log.error(\"Received an UPDATE QueryChange for Rya Instance: \" + ryaInstance +\n+                                \", Query ID: \" + change.getQueryId() + \", but the QueryRepository did not supply a \" +\n+                                \"StreamsQuery representing the created query. The query will not be processed.\");\n+                    }\n+                    break;\n+            }\n+        }\n+    }\n+\n+    /**\n+     * Processes a work queue of {@link QueryEvent}s.\n+     * <p/>\n+     * Each type of event maps the to corresponding method on {@link QueryExecutor} that is called into.\n+     */\n+    @DefaultAnnotation(NonNull.class)\n+    static class QueryEventWorker implements Runnable {\n+\n+        private final BlockingQueue<QueryEvent> workQueue;\n+        private final QueryExecutor queryExecutor;\n+        private final long pollingValue;\n+        private final TimeUnit pollingUnits;\n+        private final AtomicBoolean shutdownSignal;\n+\n+        /**\n+         * Constructs an instance of {@link QueryEventWorker}.\n+         *\n+         * @param logWorkQueue - A queue of {@link QueryEvent}s that will be worked by this object. (not null)","path":"extras/rya.streams/query-manager/src/main/java/org/apache/rya/streams/querymanager/QueryManager.java","position":814,"original_position":814,"commit_id":"50a81217e5b16a4b1377f199aefc3a1c5a225216","original_commit_id":"50a81217e5b16a4b1377f199aefc3a1c5a225216","user":{"login":"kchilton2","id":16141852,"avatar_url":"https://avatars0.githubusercontent.com/u/16141852?v=4","gravatar_id":"","url":"https://api.github.com/users/kchilton2","html_url":"https://github.com/kchilton2","followers_url":"https://api.github.com/users/kchilton2/followers","following_url":"https://api.github.com/users/kchilton2/following{/other_user}","gists_url":"https://api.github.com/users/kchilton2/gists{/gist_id}","starred_url":"https://api.github.com/users/kchilton2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kchilton2/subscriptions","organizations_url":"https://api.github.com/users/kchilton2/orgs","repos_url":"https://api.github.com/users/kchilton2/repos","events_url":"https://api.github.com/users/kchilton2/events{/privacy}","received_events_url":"https://api.github.com/users/kchilton2/received_events","type":"User","site_admin":false},"body":"Done.","created_at":"2018-02-14T19:03:01Z","updated_at":"2018-02-14T19:03:01Z","html_url":"https://github.com/apache/incubator-rya/pull/272#discussion_r168276972","pull_request_url":"https://api.github.com/repos/apache/incubator-rya/pulls/272","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/comments/168276972"},"html":{"href":"https://github.com/apache/incubator-rya/pull/272#discussion_r168276972"},"pull_request":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/272"}},"in_reply_to_id":167242070},"pull_request":{"url":"https://api.github.com/repos/apache/incubator-rya/pulls/272","id":168089046,"html_url":"https://github.com/apache/incubator-rya/pull/272","diff_url":"https://github.com/apache/incubator-rya/pull/272.diff","patch_url":"https://github.com/apache/incubator-rya/pull/272.patch","issue_url":"https://api.github.com/repos/apache/incubator-rya/issues/272","number":272,"state":"open","locked":false,"title":"RYA-443 Rya Streams Query Manager daemon program.","user":{"login":"kchilton2","id":16141852,"avatar_url":"https://avatars0.githubusercontent.com/u/16141852?v=4","gravatar_id":"","url":"https://api.github.com/users/kchilton2","html_url":"https://github.com/kchilton2","followers_url":"https://api.github.com/users/kchilton2/followers","following_url":"https://api.github.com/users/kchilton2/following{/other_user}","gists_url":"https://api.github.com/users/kchilton2/gists{/gist_id}","starred_url":"https://api.github.com/users/kchilton2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kchilton2/subscriptions","organizations_url":"https://api.github.com/users/kchilton2/orgs","repos_url":"https://api.github.com/users/kchilton2/repos","events_url":"https://api.github.com/users/kchilton2/events{/privacy}","received_events_url":"https://api.github.com/users/kchilton2/received_events","type":"User","site_admin":false},"body":"## Description\r\nImplemented an RPM installed CentOS 7 daemon service used to execute Kafka Streams jobs on the local machine. This is part of Rya Streams.\r\n\r\n### Tests\r\nWrote automated tests and executed the application on a CentOS 7 machine using the RPM.\r\n\r\n### Links\r\nhttps://issues.apache.org/jira/browse/RYA-443","created_at":"2018-02-08T21:15:46Z","updated_at":"2018-02-14T19:03:01Z","closed_at":null,"merged_at":null,"merge_commit_sha":"46000c967a88c9023bb9476c68eeb0fe8689e4c4","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/incubator-rya/pulls/272/commits","review_comments_url":"https://api.github.com/repos/apache/incubator-rya/pulls/272/comments","review_comment_url":"https://api.github.com/repos/apache/incubator-rya/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/incubator-rya/issues/272/comments","statuses_url":"https://api.github.com/repos/apache/incubator-rya/statuses/50a81217e5b16a4b1377f199aefc3a1c5a225216","head":{"label":"kchilton2:RYA-443_rya-streams-querymanager","ref":"RYA-443_rya-streams-querymanager","sha":"50a81217e5b16a4b1377f199aefc3a1c5a225216","user":{"login":"kchilton2","id":16141852,"avatar_url":"https://avatars0.githubusercontent.com/u/16141852?v=4","gravatar_id":"","url":"https://api.github.com/users/kchilton2","html_url":"https://github.com/kchilton2","followers_url":"https://api.github.com/users/kchilton2/followers","following_url":"https://api.github.com/users/kchilton2/following{/other_user}","gists_url":"https://api.github.com/users/kchilton2/gists{/gist_id}","starred_url":"https://api.github.com/users/kchilton2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kchilton2/subscriptions","organizations_url":"https://api.github.com/users/kchilton2/orgs","repos_url":"https://api.github.com/users/kchilton2/repos","events_url":"https://api.github.com/users/kchilton2/events{/privacy}","received_events_url":"https://api.github.com/users/kchilton2/received_events","type":"User","site_admin":false},"repo":{"id":54225439,"name":"incubator-rya","full_name":"kchilton2/incubator-rya","owner":{"login":"kchilton2","id":16141852,"avatar_url":"https://avatars0.githubusercontent.com/u/16141852?v=4","gravatar_id":"","url":"https://api.github.com/users/kchilton2","html_url":"https://github.com/kchilton2","followers_url":"https://api.github.com/users/kchilton2/followers","following_url":"https://api.github.com/users/kchilton2/following{/other_user}","gists_url":"https://api.github.com/users/kchilton2/gists{/gist_id}","starred_url":"https://api.github.com/users/kchilton2/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kchilton2/subscriptions","organizations_url":"https://api.github.com/users/kchilton2/orgs","repos_url":"https://api.github.com/users/kchilton2/repos","events_url":"https://api.github.com/users/kchilton2/events{/privacy}","received_events_url":"https://api.github.com/users/kchilton2/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/kchilton2/incubator-rya","description":"Mirror of Apache Rya (Incubating)","fork":true,"url":"https://api.github.com/repos/kchilton2/incubator-rya","forks_url":"https://api.github.com/repos/kchilton2/incubator-rya/forks","keys_url":"https://api.github.com/repos/kchilton2/incubator-rya/keys{/key_id}","collaborators_url":"https://api.github.com/repos/kchilton2/incubator-rya/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/kchilton2/incubator-rya/teams","hooks_url":"https://api.github.com/repos/kchilton2/incubator-rya/hooks","issue_events_url":"https://api.github.com/repos/kchilton2/incubator-rya/issues/events{/number}","events_url":"https://api.github.com/repos/kchilton2/incubator-rya/events","assignees_url":"https://api.github.com/repos/kchilton2/incubator-rya/assignees{/user}","branches_url":"https://api.github.com/repos/kchilton2/incubator-rya/branches{/branch}","tags_url":"https://api.github.com/repos/kchilton2/incubator-rya/tags","blobs_url":"https://api.github.com/repos/kchilton2/incubator-rya/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/kchilton2/incubator-rya/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/kchilton2/incubator-rya/git/refs{/sha}","trees_url":"https://api.github.com/repos/kchilton2/incubator-rya/git/trees{/sha}","statuses_url":"https://api.github.com/repos/kchilton2/incubator-rya/statuses/{sha}","languages_url":"https://api.github.com/repos/kchilton2/incubator-rya/languages","stargazers_url":"https://api.github.com/repos/kchilton2/incubator-rya/stargazers","contributors_url":"https://api.github.com/repos/kchilton2/incubator-rya/contributors","subscribers_url":"https://api.github.com/repos/kchilton2/incubator-rya/subscribers","subscription_url":"https://api.github.com/repos/kchilton2/incubator-rya/subscription","commits_url":"https://api.github.com/repos/kchilton2/incubator-rya/commits{/sha}","git_commits_url":"https://api.github.com/repos/kchilton2/incubator-rya/git/commits{/sha}","comments_url":"https://api.github.com/repos/kchilton2/incubator-rya/comments{/number}","issue_comment_url":"https://api.github.com/repos/kchilton2/incubator-rya/issues/comments{/number}","contents_url":"https://api.github.com/repos/kchilton2/incubator-rya/contents/{+path}","compare_url":"https://api.github.com/repos/kchilton2/incubator-rya/compare/{base}...{head}","merges_url":"https://api.github.com/repos/kchilton2/incubator-rya/merges","archive_url":"https://api.github.com/repos/kchilton2/incubator-rya/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/kchilton2/incubator-rya/downloads","issues_url":"https://api.github.com/repos/kchilton2/incubator-rya/issues{/number}","pulls_url":"https://api.github.com/repos/kchilton2/incubator-rya/pulls{/number}","milestones_url":"https://api.github.com/repos/kchilton2/incubator-rya/milestones{/number}","notifications_url":"https://api.github.com/repos/kchilton2/incubator-rya/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/kchilton2/incubator-rya/labels{/name}","releases_url":"https://api.github.com/repos/kchilton2/incubator-rya/releases{/id}","deployments_url":"https://api.github.com/repos/kchilton2/incubator-rya/deployments","created_at":"2016-03-18T19:08:55Z","updated_at":"2016-03-18T19:08:57Z","pushed_at":"2018-02-14T18:50:11Z","git_url":"git://github.com/kchilton2/incubator-rya.git","ssh_url":"git@github.com:kchilton2/incubator-rya.git","clone_url":"https://github.com/kchilton2/incubator-rya.git","svn_url":"https://github.com/kchilton2/incubator-rya","homepage":null,"size":7178,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"cb096b180d124987f7f85f47866f976f7c960bea","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":43734923,"name":"incubator-rya","full_name":"apache/incubator-rya","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/incubator-rya","description":"Mirror of Apache Rya (Incubating)","fork":false,"url":"https://api.github.com/repos/apache/incubator-rya","forks_url":"https://api.github.com/repos/apache/incubator-rya/forks","keys_url":"https://api.github.com/repos/apache/incubator-rya/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/incubator-rya/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/incubator-rya/teams","hooks_url":"https://api.github.com/repos/apache/incubator-rya/hooks","issue_events_url":"https://api.github.com/repos/apache/incubator-rya/issues/events{/number}","events_url":"https://api.github.com/repos/apache/incubator-rya/events","assignees_url":"https://api.github.com/repos/apache/incubator-rya/assignees{/user}","branches_url":"https://api.github.com/repos/apache/incubator-rya/branches{/branch}","tags_url":"https://api.github.com/repos/apache/incubator-rya/tags","blobs_url":"https://api.github.com/repos/apache/incubator-rya/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/incubator-rya/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/incubator-rya/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/incubator-rya/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/incubator-rya/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/incubator-rya/languages","stargazers_url":"https://api.github.com/repos/apache/incubator-rya/stargazers","contributors_url":"https://api.github.com/repos/apache/incubator-rya/contributors","subscribers_url":"https://api.github.com/repos/apache/incubator-rya/subscribers","subscription_url":"https://api.github.com/repos/apache/incubator-rya/subscription","commits_url":"https://api.github.com/repos/apache/incubator-rya/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/incubator-rya/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/incubator-rya/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/incubator-rya/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/incubator-rya/contents/{+path}","compare_url":"https://api.github.com/repos/apache/incubator-rya/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/incubator-rya/merges","archive_url":"https://api.github.com/repos/apache/incubator-rya/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/incubator-rya/downloads","issues_url":"https://api.github.com/repos/apache/incubator-rya/issues{/number}","pulls_url":"https://api.github.com/repos/apache/incubator-rya/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/incubator-rya/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/incubator-rya/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/incubator-rya/labels{/name}","releases_url":"https://api.github.com/repos/apache/incubator-rya/releases{/id}","deployments_url":"https://api.github.com/repos/apache/incubator-rya/deployments","created_at":"2015-10-06T07:00:05Z","updated_at":"2018-01-30T12:28:08Z","pushed_at":"2018-02-14T15:31:37Z","git_url":"git://github.com/apache/incubator-rya.git","ssh_url":"git@github.com:apache/incubator-rya.git","clone_url":"https://github.com/apache/incubator-rya.git","svn_url":"https://github.com/apache/incubator-rya","homepage":null,"size":6853,"stargazers_count":44,"watchers_count":44,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":50,"mirror_url":null,"archived":false,"open_issues_count":9,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0"},"forks":50,"open_issues":9,"watchers":44,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/272"},"html":{"href":"https://github.com/apache/incubator-rya/pull/272"},"issue":{"href":"https://api.github.com/repos/apache/incubator-rya/issues/272"},"comments":{"href":"https://api.github.com/repos/apache/incubator-rya/issues/272/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/272/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/incubator-rya/pulls/272/commits"},"statuses":{"href":"https://api.github.com/repos/apache/incubator-rya/statuses/50a81217e5b16a4b1377f199aefc3a1c5a225216"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"kchilton2"}}