You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2015/03/04 21:51:01 UTC
[1/2] incubator-nifi git commit: NIFI-250: Creating
ReportingTaskProvider
Repository: incubator-nifi
Updated Branches:
refs/heads/NIFI-250 2d4aebf33 -> 883c4ac9d
NIFI-250: Creating ReportingTaskProvider
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/83eff8d6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/83eff8d6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/83eff8d6
Branch: refs/heads/NIFI-250
Commit: 83eff8d6df7b485a4d0eb03f176e4a28ff81df3b
Parents: a227fe4
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Mar 4 15:50:30 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Mar 4 15:50:30 2015 -0500
----------------------------------------------------------------------
.../cluster/manager/impl/WebClusterManager.java | 27 +++++++--
.../reporting/ReportingTaskProvider.java | 61 ++++++++++++++++++++
.../apache/nifi/controller/FlowController.java | 4 +-
3 files changed, 86 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/83eff8d6/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
index f4c86f6..f3dd0a0 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
@@ -41,7 +41,9 @@ import java.util.TimerTask;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CompletionService;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -66,6 +68,7 @@ import javax.xml.transform.stream.StreamResult;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.admin.service.AuditService;
+import org.apache.nifi.annotation.lifecycle.OnAdded;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.cluster.BulletinsPayload;
import org.apache.nifi.cluster.ClusterNodeInformation;
@@ -128,8 +131,10 @@ import org.apache.nifi.controller.Heartbeater;
import org.apache.nifi.controller.ReportingTaskNode;
import org.apache.nifi.controller.StandardFlowSerializer;
import org.apache.nifi.controller.ValidationContextFactory;
+import org.apache.nifi.controller.exception.ProcessorLifeCycleException;
import org.apache.nifi.controller.reporting.ClusteredReportingTaskNode;
import org.apache.nifi.controller.reporting.ReportingTaskInstantiationException;
+import org.apache.nifi.controller.reporting.ReportingTaskProvider;
import org.apache.nifi.controller.reporting.StandardReportingInitializationContext;
import org.apache.nifi.controller.scheduling.QuartzSchedulingAgent;
import org.apache.nifi.controller.scheduling.StandardProcessScheduler;
@@ -233,7 +238,7 @@ import com.sun.jersey.api.client.ClientResponse;
*
* @author unattributed
*/
-public class WebClusterManager implements HttpClusterManager, ProtocolHandler, ControllerServiceProvider {
+public class WebClusterManager implements HttpClusterManager, ProtocolHandler, ControllerServiceProvider, ReportingTaskProvider {
public static final String ROOT_GROUP_ID_ALIAS = "root";
public static final String BULLETIN_CATEGORY = "Clustering";
@@ -315,7 +320,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
private final ClusterManagerLock writeLock = new ClusterManagerLock(resourceRWLock.writeLock(), "Write");
private final Set<Node> nodes = new HashSet<>();
- private final Map<String, ReportingTaskNode> reportingTasks = new HashMap<>();
+ private final ConcurrentMap<String, ReportingTaskNode> reportingTasks = new ConcurrentHashMap<>();
// null means the dataflow should be read from disk
private StandardDataFlow cachedDataFlow = null;
@@ -471,7 +476,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
// Load and start running Reporting Tasks
final byte[] serializedReportingTasks = clusterDataFlow.getReportingTasks();
if ( serializedReportingTasks != null && serializedReportingTasks.length > 0 ) {
- reportingTasks.putAll(loadReportingTasks(serializedReportingTasks));
+ loadReportingTasks(serializedReportingTasks);
}
} catch (final IOException ioe) {
logger.warn("Failed to initialize cluster services due to: " + ioe, ioe);
@@ -931,7 +936,7 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
//set the class to be used for the configured reporting task
final ReportingTaskNode reportingTaskNode;
try {
- reportingTaskNode = createReportingTask(taskClass, taskId);
+ reportingTaskNode = createReportingTask(taskClass, taskId, false);
} catch (final ReportingTaskInstantiationException e) {
logger.error("Unable to load reporting task {} due to {}", new Object[]{taskId, e});
if (logger.isDebugEnabled()) {
@@ -973,7 +978,9 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return tasks;
}
- private ReportingTaskNode createReportingTask(final String type, final String id) throws ReportingTaskInstantiationException {
+
+ @Override
+ public ReportingTaskNode createReportingTask(final String type, final String id, final boolean firstTimeAdded) throws ReportingTaskInstantiationException {
if (type == null) {
throw new NullPointerException();
}
@@ -1003,6 +1010,16 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this);
final ReportingTaskNode taskNode = new ClusteredReportingTaskNode(task, id, processScheduler,
new ClusteredEventAccess(this), bulletinRepository, controllerServiceProvider, validationContextFactory);
+
+ reportingTasks.put(id, taskNode);
+ if ( firstTimeAdded ) {
+ try (final NarCloseable x = NarCloseable.withNarLoader()) {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnAdded.class, task);
+ } catch (final Exception e) {
+ throw new ProcessorLifeCycleException("Failed to invoke On-Added Lifecycle methods of " + task, e);
+ }
+ }
+
return taskNode;
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/83eff8d6/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/reporting/ReportingTaskProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/reporting/ReportingTaskProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/reporting/ReportingTaskProvider.java
new file mode 100644
index 0000000..c58115f
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/reporting/ReportingTaskProvider.java
@@ -0,0 +1,61 @@
+/*
+ * 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.nifi.controller.reporting;
+
+import org.apache.nifi.controller.ReportingTaskNode;
+
+/**
+ * A ReportingTaskProvider is responsible for providing management of, and access to, Reporting Tasks
+ */
+public interface ReportingTaskProvider {
+
+ /**
+ * Creates a new instance of a reporting task
+ *
+ * @param type the type (fully qualified class name) of the reporting task to instantiate
+ * @param id the identifier for the Reporting Task
+ * @param firstTimeAdded whether or not this is the first time that the reporting task is being added
+ * to the flow. I.e., this will be true only when the user adds the reporting task to the flow, not when
+ * the flow is being restored after a restart of the software
+ *
+ * @return the ReportingTaskNode that is used to manage the reporting task
+ *
+ * @throws ReportingTaskInstantiationException if unable to create the Reporting Task
+ */
+ ReportingTaskNode createReportingTask(String type, String id, boolean firstTimeAdded) throws ReportingTaskInstantiationException;
+
+ /**
+ * Returns the reporting task that has the given identifier, or <code>null</code> if no reporting task
+ * exists with that ID.
+ *
+ * @param identifier
+ * @return
+ */
+ ReportingTaskNode getReportingTaskNode(String identifier);
+
+
+ /**
+ * Removes the given reporting task from the flow
+ *
+ * @param reportingTask
+ *
+ * @throws IllegalStateException if the reporting task cannot be removed because it is not stopped, or
+ * if the reporting task is not known in the flow
+ */
+ void removeReportingTask(ReportingTaskNode reportingTask);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/83eff8d6/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 0a86145..f6e9aae 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -78,6 +78,7 @@ import org.apache.nifi.controller.exception.ProcessorLifeCycleException;
import org.apache.nifi.controller.label.Label;
import org.apache.nifi.controller.label.StandardLabel;
import org.apache.nifi.controller.reporting.ReportingTaskInstantiationException;
+import org.apache.nifi.controller.reporting.ReportingTaskProvider;
import org.apache.nifi.controller.reporting.StandardReportingTaskNode;
import org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.CounterRepository;
@@ -190,7 +191,7 @@ import org.slf4j.LoggerFactory;
import com.sun.jersey.api.client.ClientHandlerException;
-public class FlowController implements EventAccess, ControllerServiceProvider, Heartbeater, QueueProvider {
+public class FlowController implements EventAccess, ControllerServiceProvider, ReportingTaskProvider, Heartbeater, QueueProvider {
// default repository implementations
public static final String DEFAULT_FLOWFILE_REPO_IMPLEMENTATION = "org.apache.nifi.controller.repository.WriteAheadFlowFileRepository";
@@ -2472,6 +2473,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
return createReportingTask(type, UUID.randomUUID().toString(), firstTimeAdded);
}
+ @Override
public ReportingTaskNode createReportingTask(final String type, final String id, final boolean firstTimeAdded) throws ReportingTaskInstantiationException {
if (type == null || id == null) {
throw new NullPointerException();
[2/2] incubator-nifi git commit: Merge branch 'NIFI-250' of
http://git-wip-us.apache.org/repos/asf/incubator-nifi into NIFI-250
Posted by ma...@apache.org.
Merge branch 'NIFI-250' of http://git-wip-us.apache.org/repos/asf/incubator-nifi into NIFI-250
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/883c4ac9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/883c4ac9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/883c4ac9
Branch: refs/heads/NIFI-250
Commit: 883c4ac9d7a1cdbd6989fb1312624a752c490461
Parents: 83eff8d 2d4aebf
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Mar 4 15:50:58 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Mar 4 15:50:58 2015 -0500
----------------------------------------------------------------------
.../nifi/admin/dao/impl/StandardActionDAO.java | 4 +-
...ontrollerServiceReferencingComponentDTO.java | 26 +-
.../nifi/web/api/dto/ReportingTaskDTO.java | 27 +-
.../nifi/controller/FlowFromDOMFactory.java | 2 +-
.../controller/StandardFlowSynchronizer.java | 14 +-
.../java/org/apache/nifi/action/Component.java | 3 +-
.../nifi/audit/ControllerServiceAuditor.java | 27 +-
.../apache/nifi/audit/ReportingTaskAuditor.java | 353 +++++++++++
.../org/apache/nifi/web/NiFiServiceFacade.java | 21 +-
.../nifi/web/StandardNiFiServiceFacade.java | 11 +-
.../apache/nifi/web/api/ControllerResource.java | 12 +-
.../nifi/web/api/ControllerServiceResource.java | 52 +-
.../apache/nifi/web/api/HistoryResource.java | 31 +
.../org/apache/nifi/web/api/dto/DtoFactory.java | 77 ++-
.../nifi/web/dao/ControllerServiceDAO.java | 17 +-
.../dao/impl/StandardControllerServiceDAO.java | 91 +--
.../web/dao/impl/StandardReportingTaskDAO.java | 3 +-
.../src/main/resources/nifi-web-api-context.xml | 17 +-
.../nifi-framework/nifi-web/nifi-web-ui/pom.xml | 3 +
.../main/resources/filters/canvas.properties | 1 +
.../src/main/webapp/WEB-INF/pages/canvas.jsp | 2 +
.../canvas/controller-service-configuration.jsp | 2 +-
.../canvas/new-reporting-task-dialog.jsp | 53 ++
.../canvas/reporting-task-configuration.jsp | 63 ++
.../nifi-web-ui/src/main/webapp/css/canvas.css | 2 +
.../src/main/webapp/css/controller-service.css | 7 -
.../webapp/css/new-reporting-task-dialog.css | 152 +++++
.../src/main/webapp/css/reporting-task.css | 80 +++
.../src/main/webapp/js/nf/canvas/nf-canvas.js | 1 +
.../js/nf/canvas/nf-controller-service.js | 240 ++++---
.../webapp/js/nf/canvas/nf-reporting-task.js | 426 +++++++++++++
.../src/main/webapp/js/nf/canvas/nf-settings.js | 618 +++++++++++++++++--
.../webapp/js/nf/history/nf-history-table.js | 2 +-
33 files changed, 2188 insertions(+), 252 deletions(-)
----------------------------------------------------------------------