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/01/26 21:17:25 UTC
[1/5] incubator-nifi git commit: Added Eclipse-specific files to
.gitignore
Repository: incubator-nifi
Updated Branches:
refs/heads/NIFI-250 2da579242 -> 2df4500c0
Added Eclipse-specific files to .gitignore
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/80f02e47
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/80f02e47
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/80f02e47
Branch: refs/heads/NIFI-250
Commit: 80f02e47256590c57a811b79040b70c827e71ff2
Parents: 35616e9
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jan 26 10:31:06 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jan 26 10:31:06 2015 -0500
----------------------------------------------------------------------
.gitignore | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/80f02e47/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index db4773d..e1c837c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,8 @@ target
nbactions.xml
nb-configuration.xml
.DS_Store
+.metadata
+.recommenders
# Intellij
.idea/
[3/5] incubator-nifi git commit: NIFI-250: Merged changes
Posted by ma...@apache.org.
NIFI-250: Merged changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/ff43b039
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/ff43b039
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/ff43b039
Branch: refs/heads/NIFI-250
Commit: ff43b039085c56bb5f53148e25878ad71972a3db
Parents: 7a3d208 2da5792
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jan 26 11:53:17 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jan 26 11:53:17 2015 -0500
----------------------------------------------------------------------
.../nifi/annotation/behavior/EventDriven.java | 49 ++++
.../annotation/behavior/SideEffectFree.java | 47 ++++
.../annotation/behavior/SupportsBatching.java | 52 ++++
.../annotation/behavior/TriggerSerially.java | 40 +++
.../TriggerWhenAnyDestinationAvailable.java | 42 +++
.../annotation/behavior/TriggerWhenEmpty.java | 42 +++
.../documentation/CapabilityDescription.java | 41 +++
.../nifi/annotation/documentation/Tags.java | 46 +++
.../nifi/annotation/lifecycle/OnAdded.java | 44 +++
.../nifi/annotation/lifecycle/OnDisabled.java | 46 +++
.../nifi/annotation/lifecycle/OnEnabled.java | 64 +++++
.../nifi/annotation/lifecycle/OnRemoved.java | 43 +++
.../nifi/annotation/lifecycle/OnScheduled.java | 62 +++++
.../nifi/annotation/lifecycle/OnShutdown.java | 42 +++
.../nifi/annotation/lifecycle/OnStopped.java | 59 ++++
.../annotation/lifecycle/OnUnscheduled.java | 58 ++++
.../controller/annotation/OnConfigured.java | 3 +
.../AbstractSessionFactoryProcessor.java | 4 +-
.../annotation/CapabilityDescription.java | 3 +
.../nifi/processor/annotation/EventDriven.java | 3 +
.../nifi/processor/annotation/OnAdded.java | 2 +
.../nifi/processor/annotation/OnRemoved.java | 2 +
.../nifi/processor/annotation/OnScheduled.java | 4 +-
.../nifi/processor/annotation/OnShutdown.java | 2 +
.../nifi/processor/annotation/OnStopped.java | 2 +
.../processor/annotation/OnUnscheduled.java | 2 +
.../processor/annotation/SideEffectFree.java | 2 +
.../processor/annotation/SupportsBatching.java | 2 +
.../apache/nifi/processor/annotation/Tags.java | 3 +
.../processor/annotation/TriggerSerially.java | 4 +-
.../TriggerWhenAnyDestinationAvailable.java | 2 +
.../processor/annotation/TriggerWhenEmpty.java | 2 +
.../util/ControllerServiceConfiguration.java | 2 +-
.../nifi/util/MockControllerServiceLookup.java | 10 +
.../apache/nifi/util/MockProcessContext.java | 1 +
.../apache/nifi/util/MockProcessSession.java | 1 -
.../nifi/util/StandardProcessorTestRunner.java | 231 ++++++++++++++-
.../java/org/apache/nifi/util/TestRunner.java | 155 +++++++++++
.../util/TestStandardProcessorTestRunner.java | 86 ++++++
.../cluster/manager/impl/WebClusterManager.java | 26 +-
.../nifi/controller/ReportingTaskNode.java | 21 ++
.../service/ControllerServiceNode.java | 9 +-
.../service/ControllerServiceProvider.java | 32 ++-
.../apache/nifi/controller/FlowController.java | 70 ++++-
.../nifi/controller/StandardProcessorNode.java | 46 +--
.../reporting/AbstractReportingTaskNode.java | 99 +++++++
.../scheduling/EventDrivenSchedulingAgent.java | 6 +-
.../scheduling/StandardProcessScheduler.java | 120 +++++---
.../service/ControllerServiceLoader.java | 19 +-
.../service/StandardControllerServiceNode.java | 85 +++++-
.../StandardControllerServiceProvider.java | 65 ++++-
.../tasks/ContinuallyRunConnectableTask.java | 6 +-
.../tasks/ContinuallyRunProcessorTask.java | 6 +-
.../controller/tasks/ReportingTaskWrapper.java | 6 +-
.../nifi/groups/StandardProcessGroup.java | 11 +-
.../processor/StandardSchedulingContext.java | 4 +-
.../org/apache/nifi/util/ReflectionUtils.java | 278 ++++++++++++++-----
.../StubAttributeLoggerProcessor.java | 2 +-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 28 +-
.../nifi/web/controller/ControllerFacade.java | 18 +-
.../dao/impl/StandardControllerServiceDAO.java | 9 +-
.../webapp/js/nf/canvas/nf-canvas-header.js | 4 +-
.../src/main/webapp/js/nf/canvas/nf-settings.js | 102 ++++++-
.../hadoop/AbstractHadoopProcessor.java | 5 +-
.../hadoop/CreateHadoopSequenceFile.java | 7 +-
.../apache/nifi/processors/hadoop/GetHDFS.java | 9 +-
.../processors/hadoop/GetHDFSSequenceFile.java | 7 +-
.../apache/nifi/processors/hadoop/PutHDFS.java | 6 +-
.../apache/nifi/processors/kafka/GetKafka.java | 12 +-
.../apache/nifi/processors/kafka/PutKafka.java | 8 +-
.../nifi/processors/kafka/TestPutKafka.java | 2 +-
.../standard/Base64EncodeContent.java | 41 +--
.../processors/standard/CompressContent.java | 10 +-
.../nifi/processors/standard/ControlRate.java | 8 +-
.../standard/ConvertCharacterSet.java | 10 +-
.../processors/standard/DetectDuplicate.java | 8 +-
.../processors/standard/DistributeLoad.java | 14 +-
.../processors/standard/EncryptContent.java | 10 +-
.../standard/EvaluateRegularExpression.java | 10 +-
.../nifi/processors/standard/EvaluateXPath.java | 12 +-
.../processors/standard/EvaluateXQuery.java | 10 +-
.../standard/ExecuteStreamCommand.java | 8 +-
.../processors/standard/GenerateFlowFile.java | 8 +-
.../apache/nifi/processors/standard/GetFTP.java | 6 +-
.../nifi/processors/standard/GetFile.java | 8 +-
.../processors/standard/GetFileTransfer.java | 2 +-
.../nifi/processors/standard/GetHTTP.java | 6 +-
.../nifi/processors/standard/GetJMSQueue.java | 8 +-
.../nifi/processors/standard/GetJMSTopic.java | 14 +-
.../nifi/processors/standard/GetSFTP.java | 6 +-
.../nifi/processors/standard/HashAttribute.java | 10 +-
.../nifi/processors/standard/HashContent.java | 8 +-
.../processors/standard/IdentifyMimeType.java | 10 +-
.../nifi/processors/standard/InvokeHTTP.java | 6 +-
.../nifi/processors/standard/ListenHTTP.java | 8 +-
.../nifi/processors/standard/ListenUDP.java | 12 +-
.../nifi/processors/standard/LogAttribute.java | 8 +-
.../nifi/processors/standard/MergeContent.java | 12 +-
.../nifi/processors/standard/ModifyBytes.java | 8 +-
.../processors/standard/MonitorActivity.java | 10 +-
.../nifi/processors/standard/PostHTTP.java | 10 +-
.../nifi/processors/standard/PutEmail.java | 6 +-
.../apache/nifi/processors/standard/PutFTP.java | 8 +-
.../nifi/processors/standard/PutFile.java | 6 +-
.../apache/nifi/processors/standard/PutJMS.java | 6 +-
.../nifi/processors/standard/PutSFTP.java | 6 +-
.../nifi/processors/standard/ReplaceText.java | 10 +-
.../standard/ReplaceTextWithMapping.java | 10 +-
.../processors/standard/RouteOnAttribute.java | 10 +-
.../processors/standard/RouteOnContent.java | 10 +-
.../nifi/processors/standard/ScanAttribute.java | 12 +-
.../nifi/processors/standard/ScanContent.java | 10 +-
.../processors/standard/SegmentContent.java | 10 +-
.../nifi/processors/standard/SplitContent.java | 10 +-
.../nifi/processors/standard/SplitText.java | 10 +-
.../nifi/processors/standard/SplitXml.java | 10 +-
.../nifi/processors/standard/TransformXml.java | 10 +-
.../nifi/processors/standard/UnpackContent.java | 10 +-
.../nifi/processors/standard/ValidateXml.java | 12 +-
.../standard/RESTServiceContentModified.java | 2 +-
.../standard/TestCompressContent.java | 5 +-
.../standard/TestConvertCharacterSet.java | 3 -
.../standard/TestDetectDuplicate.java | 10 +-
.../nifi/processors/standard/TestGetHTTP.java | 16 +-
.../processors/standard/TestInvokeHTTP.java | 4 +-
.../ControllerStatusReportingTask.java | 13 +-
.../nifi/controller/MonitorDiskUsage.java | 7 +-
.../apache/nifi/controller/MonitorMemory.java | 12 +-
.../ganglia/StandardGangliaReporter.java | 15 +-
.../cache/client/DistributedMapCacheClient.java | 5 +
.../cache/client/DistributedSetCacheClient.java | 5 +
.../DistributedMapCacheClientService.java | 9 +-
.../DistributedSetCacheClientService.java | 9 +-
.../cache/server/DistributedCacheServer.java | 6 +-
.../cache/server/DistributedSetCacheServer.java | 6 +-
.../server/map/DistributedMapCacheServer.java | 5 +
.../nifi/ssl/StandardSSLContextService.java | 9 +-
.../apache/nifi/ssl/SSLContextServiceTest.java | 182 +++++-------
.../org/apache/nifi/ssl/SSLContextService.java | 5 +
.../processors/attributes/UpdateAttribute.java | 11 +-
140 files changed, 2583 insertions(+), 656 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
index 17a6452,656759a..5f1c460
--- 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
@@@ -63,6 -63,6 +63,7 @@@ import javax.xml.validation.Schema
import javax.xml.validation.SchemaFactory;
import javax.xml.validation.Validator;
++import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.admin.service.AuditService;
import org.apache.nifi.cluster.BulletinsPayload;
import org.apache.nifi.cluster.ClusterNodeInformation;
@@@ -117,6 -117,6 +118,7 @@@ import org.apache.nifi.cluster.protocol
import org.apache.nifi.cluster.protocol.message.PrimaryRoleAssignmentMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage.MessageType;
++import org.apache.nifi.cluster.protocol.message.ReconnectionFailureMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionRequestMessage;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ControllerService;
@@@ -191,7 -191,7 +193,6 @@@ import org.apache.nifi.web.api.entity.P
import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupsEntity;
import org.apache.nifi.web.util.WebUtils;
--import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.DOMException;
@@@ -203,8 -203,8 +204,6 @@@ import org.xml.sax.SAXParseException
import com.sun.jersey.api.client.ClientResponse;
--import org.apache.nifi.cluster.protocol.message.ReconnectionFailureMessage;
--
/**
* Provides a cluster manager implementation. The manager federates incoming
* HTTP client requests to the nodes' external API using the HTTP protocol. The
@@@ -1289,18 -1289,6 +1288,18 @@@ public class WebClusterManager implemen
}
}
+ /**
+ * Adds an instance of a specified controller service.
+ *
+ * @param type
+ * @param id
+ * @param properties
+ * @return
+ */
+ @Override
- public ControllerServiceNode createControllerService(String type) {
- return controllerServiceProvider.createControllerService(type);
++ public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
++ return controllerServiceProvider.createControllerService(type, firstTimeAdded);
+ }
@Override
public ControllerService getControllerService(String serviceIdentifier) {
@@@ -1323,10 -1311,27 +1322,27 @@@
}
@Override
- public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
- return controllerServiceProvider.createControllerService(type, id, firstTimeAdded);
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return controllerServiceProvider.getControllerServiceName(serviceIdentifier);
}
-
+
+ @Override
+ public void removeControllerService(final ControllerServiceNode serviceNode) {
+ controllerServiceProvider.removeControllerService(serviceNode);
+ }
+
+
+ @Override
+ public void enableControllerService(final ControllerServiceNode serviceNode) {
+ controllerServiceProvider.enableControllerService(serviceNode);
+ }
+
+ @Override
+ public void disableControllerService(final ControllerServiceNode serviceNode) {
+ controllerServiceProvider.disableControllerService(serviceNode);
+ }
+
+
/**
* Handle a bulletins message.
*
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
index ba00799,5f2fc2e..54a506e
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
@@@ -24,14 -25,15 +25,14 @@@ import org.apache.nifi.controller.Contr
public interface ControllerServiceProvider extends ControllerServiceLookup {
/**
- * Creates a new Controller Service of the specified type
- * Creates a new Controller Service of the given type and assigns it the given id. If <code>firstTimeadded</code>
++ * Creates a new Controller Service of the specified type and assigns it the given id. If <code>firstTimeadded</code>
+ * is true, calls any methods that are annotated with {@link OnAdded}
*
* @param type
-- * @param id
- * @param properties
+ * @param firstTimeAdded
* @return
*/
- ControllerServiceNode createControllerService(String type);
- ControllerServiceNode createControllerService(String type, String id, boolean firstTimeAdded);
++ ControllerServiceNode createControllerService(String type, boolean firstTimeAdded);
/**
* Gets the controller service node for the specified identifier. Returns
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index ae0ef97,7cb2874..38c8d42
--- 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
@@@ -48,8 -48,9 +48,10 @@@ import java.util.concurrent.locks.Reent
import javax.net.ssl.SSLContext;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.admin.service.UserService;
+ 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.HeartbeatPayload;
import org.apache.nifi.cluster.protocol.DataFlow;
@@@ -2523,32 -2556,29 +2556,56 @@@ public class FlowController implements
return reportingTasks.values();
}
-
+ /**
+ * Recursively stops all Processors and Reporting Tasks that are referencing the given Controller Service,
+ * as well as disabling any Controller Service that references this Controller Service (and stops
+ * all Reporting Task or Controller Service that is referencing it, and so on).
+ * @param serviceNode
+ */
+ public void deactiveReferencingComponents(final ControllerServiceNode serviceNode) {
+ final ControllerServiceReference reference = serviceNode.getReferences();
+
+ final Set<ConfiguredComponent> components = reference.getActiveReferences();
+ for (final ConfiguredComponent component : components) {
+ if ( component instanceof ControllerServiceNode ) {
+ deactiveReferencingComponents((ControllerServiceNode) component);
+ // TODO: DISABLE CONTROLLER SERVICE!
+ } else if ( component instanceof ReportingTaskNode ) {
+ stopReportingTask((ReportingTaskNode) component);
+ } else if ( component instanceof ProcessorNode ) {
+ final ProcessorNode procNode = (ProcessorNode) component;
+ stopProcessor(procNode.getProcessGroup().getIdentifier(), procNode.getIdentifier());
+ }
+ }
+ }
+
+ @Override
- public ControllerServiceNode createControllerService(final String type) {
- return controllerServiceProvider.createControllerService(type);
++ public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
++ return controllerServiceProvider.createControllerService(type, firstTimeAdded);
++ }
++
+ public void enableReportingTask(final ReportingTaskNode reportingTaskNode) {
+ reportingTaskNode.verifyCanEnable();
+
+ processScheduler.enableReportingTask(reportingTaskNode);
+ }
+
+ public void disableReportingTask(final ReportingTaskNode reportingTaskNode) {
+ reportingTaskNode.verifyCanDisable();
+
+ processScheduler.disableReportingTask(reportingTaskNode);
+ }
+
+ @Override
+ public void enableControllerService(final ControllerServiceNode serviceNode) {
+ serviceNode.verifyCanEnable();
+ controllerServiceProvider.enableControllerService(serviceNode);
+ }
+
+ @Override
+ public void disableControllerService(final ControllerServiceNode serviceNode) {
+ serviceNode.verifyCanDisable();
+ controllerServiceProvider.disableControllerService(serviceNode);
}
@Override
@@@ -2572,10 -2607,9 +2629,13 @@@
}
@Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return controllerServiceProvider.getControllerServiceName(serviceIdentifier);
+ }
+
+ public void removeControllerService(final ControllerServiceNode serviceNode) {
+ controllerServiceProvider.removeControllerService(serviceNode);
+ }
//
// Counters
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index e3e6e6b,741caec..b829bbd
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@@ -122,4 -136,61 +136,61 @@@ public class StandardControllerServiceN
throw new IllegalStateException("Cannot modify Controller Service configuration because it is currently enabled. Please disable the Controller Service first.");
}
}
+
+ @Override
+ public void setProperty(final String name, final String value) {
+ super.setProperty(name, value);
+
+ onConfigured();
+ }
+
+ @Override
+ public boolean removeProperty(String name) {
+ final boolean removed = super.removeProperty(name);
+ if ( removed ) {
+ onConfigured();
+ }
+
+ return removed;
+ }
+
+ @SuppressWarnings("deprecation")
+ private void onConfigured() {
+ try (final NarCloseable x = NarCloseable.withNarLoader()) {
+ final ConfigurationContext configContext = new StandardConfigurationContext(this, serviceProvider);
+ ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, implementation, configContext);
+ } catch (final Exception e) {
+ throw new ProcessorLifeCycleException("Failed to invoke On-Configured Lifecycle methods of " + implementation, e);
+ }
+ }
+
+ @Override
+ public void verifyCanDelete() {
+ if ( !isDisabled() ) {
+ throw new IllegalStateException(implementation + " cannot be deleted because it is not disabled");
+ }
+ }
+
+ @Override
+ public void verifyCanDisable() {
+ final ControllerServiceReference references = getReferences();
- final int numRunning = references.getRunningReferences().size();
++ final int numRunning = references.getActiveReferences().size();
+ if ( numRunning > 0 ) {
+ throw new IllegalStateException(implementation + " cannot be disabled because it is referenced by " + numRunning + " components that are currently running");
+ }
+ }
+
+ @Override
+ public void verifyCanEnable() {
+ if ( !isDisabled() ) {
+ throw new IllegalStateException(implementation + " cannot be enabled because it is not disabled");
+ }
+ }
+
+ @Override
+ public void verifyCanUpdate() {
+ if ( !isDisabled() ) {
+ throw new IllegalStateException(implementation + " cannot be updated because it is not disabled");
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 66cc541,a40e43e..b07fc3f
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@@ -28,12 -28,18 +28,18 @@@ import java.util.HashSet
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+ import org.apache.nifi.annotation.lifecycle.OnAdded;
+ import org.apache.nifi.annotation.lifecycle.OnEnabled;
+ import org.apache.nifi.annotation.lifecycle.OnDisabled;
+ import org.apache.nifi.annotation.lifecycle.OnRemoved;
+ import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ValidationContextFactory;
-import org.apache.nifi.controller.exception.ControllerServiceAlreadyExistsException;
import org.apache.nifi.controller.exception.ControllerServiceNotFoundException;
+ import org.apache.nifi.controller.exception.ProcessorLifeCycleException;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.StandardValidationContextFactory;
@@@ -90,12 -97,14 +97,12 @@@ public class StandardControllerServiceP
}
@Override
- public ControllerServiceNode createControllerService(final String type) {
- public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
- if (type == null || id == null) {
++ public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
+ if (type == null) {
throw new NullPointerException();
}
- if (controllerServices.containsKey(id)) {
- throw new ControllerServiceAlreadyExistsException(id);
- }
-
+
+ final String id = UUID.randomUUID().toString();
final ClassLoader currentContextClassLoader = Thread.currentThread().getContextClassLoader();
try {
final ClassLoader cl = ExtensionManager.getClassLoader(type);
@@@ -134,9 -144,18 +142,17 @@@
final ValidationContextFactory validationContextFactory = new StandardValidationContextFactory(this);
- final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, id, validationContextFactory, this);
+ final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, originalService, id, validationContextFactory, this);
serviceNodeHolder.set(serviceNode);
- serviceNode.setAnnotationData(null);
- serviceNode.setName(id);
+ serviceNode.setName(rawClass.getSimpleName());
+
+ if ( firstTimeAdded ) {
+ try (final NarCloseable x = NarCloseable.withNarLoader()) {
+ ReflectionUtils.invokeMethodsWithAnnotation(OnAdded.class, originalService);
+ } catch (final Exception e) {
+ throw new ProcessorLifeCycleException("Failed to invoke On-Added Lifecycle methods of " + originalService, e);
+ }
+ }
this.controllerServices.put(id, serviceNode);
return serviceNode;
@@@ -184,8 -228,19 +225,24 @@@
}
@Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ final ControllerServiceNode node = getControllerServiceNode(serviceIdentifier);
+ return node == null ? null : node.getName();
+ }
++
+ public void removeControllerService(final ControllerServiceNode serviceNode) {
+ final ControllerServiceNode existing = controllerServices.get(serviceNode.getIdentifier());
+ if ( existing == null || existing != serviceNode ) {
+ throw new IllegalStateException("Controller Service " + serviceNode + " does not exist in this Flow");
+ }
+
+ serviceNode.verifyCanDelete();
+
+ try (final NarCloseable x = NarCloseable.withNarLoader()) {
+ final ConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this);
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class, serviceNode.getControllerServiceImplementation(), configurationContext);
+ }
+
+ controllerServices.remove(serviceNode.getIdentifier());
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index 27c60bb,2dec8fc..603c1d2
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@@ -405,8 -406,12 +405,12 @@@ public class ControllerFacade implement
}
@Override
- public ControllerServiceNode createControllerService(String type) {
- return flowController.createControllerService(type);
- public ControllerServiceNode createControllerService(String type, String id, boolean firstTimeAdded) {
- return flowController.createControllerService(type, id, firstTimeAdded);
++ public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
++ return flowController.createControllerService(type, firstTimeAdded);
+ }
+
+ public void removeControllerService(ControllerServiceNode serviceNode) {
+ flowController.removeControllerService(serviceNode);
}
@Override
@@@ -428,13 -433,17 +432,23 @@@
public boolean isControllerServiceEnabled(final String serviceIdentifier) {
return flowController.isControllerServiceEnabled(serviceIdentifier);
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return flowController.getControllerServiceName(serviceIdentifier);
+ }
+
+ @Override
+ public void enableControllerService(final ControllerServiceNode serviceNode) {
+ flowController.enableControllerService(serviceNode);
+ }
+
+ @Override
+ public void disableControllerService(ControllerServiceNode serviceNode) {
+ flowController.disableControllerService(serviceNode);
+ }
+
/**
* Gets the status of this controller.
*
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ff43b039/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
----------------------------------------------------------------------
diff --cc nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
index 6feab86,0cdde6b..116394c
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardControllerServiceDAO.java
@@@ -16,8 -16,9 +16,7 @@@
*/
package org.apache.nifi.web.dao.impl;
--import java.util.HashMap;
- import java.util.Map;
+ import java.util.HashSet;
-import java.util.Map;
import java.util.Set;
import org.apache.nifi.controller.FlowController;
@@@ -56,10 -57,8 +55,8 @@@ public class StandardControllerServiceD
*/
@Override
public ControllerServiceNode createControllerService(final ControllerServiceDTO controllerServiceDTO) {
- final Map<String, String> temp = new HashMap<>();
-
// create the controller service
- final ControllerServiceNode controllerService = flowController.createControllerService(controllerServiceDTO.getType(), controllerServiceDTO.getName(), temp);
- final ControllerServiceNode controllerService = flowController.createControllerService(controllerServiceDTO.getType(), controllerServiceDTO.getName(), true);
++ final ControllerServiceNode controllerService = flowController.createControllerService(controllerServiceDTO.getType(), true);
return controllerService;
}
[5/5] incubator-nifi git commit: NIFI-250: Implemented fingerprinting
and reloading of controller services and reporting tasks from flow.xml
Posted by ma...@apache.org.
NIFI-250: Implemented fingerprinting and reloading of controller services and reporting tasks from flow.xml
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/2df4500c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/2df4500c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/2df4500c
Branch: refs/heads/NIFI-250
Commit: 2df4500c05ac91d46614e9ca41974fed890d2646
Parents: 86d15f9
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jan 26 15:17:04 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jan 26 15:17:04 2015 -0500
----------------------------------------------------------------------
.../cluster/manager/impl/WebClusterManager.java | 8 +-
.../service/ControllerServiceProvider.java | 2 +-
.../apache/nifi/controller/FlowController.java | 55 ++++++++++++-
.../controller/StandardFlowSynchronizer.java | 70 ++++++++++++++++-
.../service/ControllerServiceLoader.java | 3 +-
.../StandardControllerServiceProvider.java | 6 +-
.../nifi/fingerprint/FingerprintFactory.java | 81 ++++++++++++++++++++
.../nifi/web/controller/ControllerFacade.java | 8 +-
8 files changed, 219 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/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 54c2b55..9d9640d 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
@@ -1287,6 +1287,10 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
writeLock.unlock("handleControllerStartupFailure");
}
}
+
+ public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
+ return createControllerService(type, UUID.randomUUID().toString(), firstTimeAdded);
+ }
/**
* Adds an instance of a specified controller service.
@@ -1297,8 +1301,8 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
* @return
*/
@Override
- public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
- return controllerServiceProvider.createControllerService(type, firstTimeAdded);
+ public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
+ return controllerServiceProvider.createControllerService(type, id, firstTimeAdded);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
index 60ff6c9..1bc3964 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
@@ -34,7 +34,7 @@ public interface ControllerServiceProvider extends ControllerServiceLookup {
* @param firstTimeAdded
* @return
*/
- ControllerServiceNode createControllerService(String type, boolean firstTimeAdded);
+ ControllerServiceNode createControllerService(String type, String id, boolean firstTimeAdded);
/**
* Gets the controller service node for the specified identifier. Returns
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/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 92975ad..47e26c0 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
@@ -2589,9 +2589,60 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
}
}
- @Override
+
+ /**
+ * <p>
+ * Starts any enabled Processors and Reporting Tasks that are referencing this Controller Service. If other Controller
+ * Services reference this Controller Service, will also enable those services and 'active' any components referencing
+ * them.
+ * </p>
+ *
+ * <p>
+ * NOTE: If any component cannot be started, an IllegalStateException will be thrown an no more components will
+ * be activated. This method provides no atomicity.
+ * </p>
+ *
+ * @param serviceNode
+ */
+ public void activateReferencingComponents(final ControllerServiceNode serviceNode) {
+ final ControllerServiceReference ref = serviceNode.getReferences();
+ final Set<ConfiguredComponent> components = ref.getReferencingComponents();
+
+ // First, activate any other controller services. We do this first so that we can
+ // avoid the situation where Processor X depends on Controller Services Y and Z; and
+ // Controller Service Y depends on Controller Service Z. In this case, if we first attempted
+ // to start Processor X, we would fail because Controller Service Y is disabled. THis way, we
+ // can recursively enable everything.
+ for ( final ConfiguredComponent component : components ) {
+ if (component instanceof ControllerServiceNode) {
+ final ControllerServiceNode componentNode = (ControllerServiceNode) component;
+ enableControllerService(componentNode);
+ activateReferencingComponents(componentNode);
+ }
+ }
+
+ for ( final ConfiguredComponent component : components ) {
+ if (component instanceof ProcessorNode) {
+ final ProcessorNode procNode = (ProcessorNode) component;
+ if ( !procNode.isRunning() ) {
+ startProcessor(procNode.getProcessGroup().getIdentifier(), procNode.getIdentifier());
+ }
+ } else if (component instanceof ReportingTaskNode) {
+ final ReportingTaskNode taskNode = (ReportingTaskNode) component;
+ if ( !taskNode.isRunning() ) {
+ startReportingTask(taskNode);
+ }
+ }
+ }
+ }
+
public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
- return controllerServiceProvider.createControllerService(type, firstTimeAdded);
+ return createControllerService(type, UUID.randomUUID().toString(), firstTimeAdded);
+ }
+
+ @Override
+ public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
+ return controllerServiceProvider.createControllerService(type, id, firstTimeAdded);
}
public void enableReportingTask(final ReportingTaskNode reportingTaskNode) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
index 0964a91..dbf9b2c 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
@@ -238,7 +238,11 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
if ( controllerServicesElement != null ) {
final List<Element> serviceElements = DomUtils.getChildElementsByTagName(controllerServicesElement, "controllerService");
for ( final Element serviceElement : serviceElements ) {
- addControllerService(controller, serviceElement, encryptor);
+ if ( !initialized || existingFlowEmpty ) {
+ addControllerService(controller, serviceElement, encryptor);
+ } else {
+ updateControllerService(controller, serviceElement, encryptor);
+ }
}
}
@@ -246,7 +250,11 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
if ( reportingTasksElement != null ) {
final List<Element> taskElements = DomUtils.getChildElementsByTagName(reportingTasksElement, "reportingTask");
for ( final Element taskElement : taskElements ) {
- addReportingTask(controller, taskElement, encryptor);
+ if ( !initialized || existingFlowEmpty ) {
+ addReportingTask(controller, taskElement, encryptor);
+ } else {
+ updateReportingTask(controller, taskElement, encryptor);
+ }
}
}
}
@@ -347,6 +355,21 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
for ( final Map.Entry<String, String> property : dto.getProperties().entrySet() ) {
node.setProperty(property.getKey(), property.getValue());
}
+
+ if ( dto.getEnabled() == Boolean.TRUE ) {
+ controller.enableControllerService(node);
+ }
+ }
+
+ private void updateControllerService(final FlowController controller, final Element controllerServiceElement, final StringEncryptor encryptor) {
+ final ControllerServiceDTO dto = FlowFromDOMFactory.getControllerService(controllerServiceElement, encryptor);
+
+ final boolean enabled = controller.isControllerServiceEnabled(dto.getId());
+ if (dto.getEnabled() && !enabled) {
+ controller.enableControllerService(controller.getControllerServiceNode(dto.getId()));
+ } else if (dto.getEnabled() == Boolean.FALSE && enabled) {
+ controller.disableControllerService(controller.getControllerServiceNode(dto.getId()));
+ }
}
private void addReportingTask(final FlowController controller, final Element reportingTaskElement, final StringEncryptor encryptor) throws ReportingTaskInstantiationException {
@@ -367,6 +390,49 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
}
}
+ private void updateReportingTask(final FlowController controller, final Element reportingTaskElement, final StringEncryptor encryptor) {
+ final ReportingTaskDTO dto = FlowFromDOMFactory.getReportingTask(reportingTaskElement, encryptor);
+ final ReportingTaskNode taskNode = controller.getReportingTaskNode(dto.getId());
+
+ if (!taskNode.getScheduledState().name().equals(dto.getScheduledState())) {
+ try {
+ switch (ScheduledState.valueOf(dto.getScheduledState())) {
+ case DISABLED:
+ if ( taskNode.isRunning() ) {
+ controller.stopReportingTask(taskNode);
+ }
+ controller.disableReportingTask(taskNode);
+ break;
+ case RUNNING:
+ if ( taskNode.getScheduledState() == ScheduledState.DISABLED ) {
+ controller.enableReportingTask(taskNode);
+ }
+ controller.startReportingTask(taskNode);
+ break;
+ case STOPPED:
+ if (taskNode.getScheduledState() == ScheduledState.DISABLED) {
+ controller.enableReportingTask(taskNode);
+ } else if (taskNode.getScheduledState() == ScheduledState.RUNNING) {
+ controller.stopReportingTask(taskNode);
+ }
+ break;
+ }
+ } catch (final IllegalStateException ise) {
+ logger.error("Failed to change Scheduled State of {} from {} to {} due to {}", taskNode, taskNode.getScheduledState().name(), dto.getScheduledState(), ise.toString());
+ logger.error("", ise);
+
+ // create bulletin for the Processor Node
+ controller.getBulletinRepository().addBulletin(BulletinFactory.createBulletin("Node Reconnection", Severity.ERROR.name(),
+ "Failed to change Scheduled State of " + taskNode + " from " + taskNode.getScheduledState().name() + " to " + dto.getScheduledState() + " due to " + ise.toString()));
+
+ // create bulletin at Controller level.
+ controller.getBulletinRepository().addBulletin(BulletinFactory.createBulletin("Node Reconnection", Severity.ERROR.name(),
+ "Failed to change Scheduled State of " + taskNode + " from " + taskNode.getScheduledState().name() + " to " + dto.getScheduledState() + " due to " + ise.toString()));
+ }
+ }
+ }
+
+
private ProcessGroup updateProcessGroup(final FlowController controller, final ProcessGroup parentGroup, final Element processGroupElement, final StringEncryptor encryptor) throws ProcessorInstantiationException {
// get the parent group ID
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
index 7504d13..db44b5f 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
@@ -122,10 +122,11 @@ public class ControllerServiceLoader {
for (final Element serviceElement : serviceNodes) {
//get properties for the specific controller task - id, name, class,
//and schedulingPeriod must be set
+ final String serviceId = DomUtils.getChild(serviceElement, "identifier").getTextContent().trim();
final String serviceClass = DomUtils.getChild(serviceElement, "class").getTextContent().trim();
//set the class to be used for the configured controller task
- final ControllerServiceNode serviceNode = provider.createControllerService(serviceClass, false);
+ final ControllerServiceNode serviceNode = provider.createControllerService(serviceClass, serviceId, false);
//optional task-specific properties
for (final Element optionalProperty : DomUtils.getChildElementsByTagName(serviceElement, "property")) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 75de4b9..0263ee0 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -28,7 +28,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -98,12 +97,11 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
}
@Override
- public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
- if (type == null) {
+ public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
+ if (type == null || id == null) {
throw new NullPointerException();
}
- final String id = UUID.randomUUID().toString();
final ClassLoader currentContextClassLoader = Thread.currentThread().getContextClassLoader();
try {
final ClassLoader cl = ExtensionManager.getClassLoader(type);
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
index 8575569..ee7cd54 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/fingerprint/FingerprintFactory.java
@@ -41,12 +41,14 @@ import javax.xml.validation.SchemaFactory;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.FlowFromDOMFactory;
import org.apache.nifi.controller.Template;
import org.apache.nifi.controller.exception.ProcessorInstantiationException;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.util.DomUtils;
import org.apache.nifi.web.api.dto.ConnectionDTO;
+import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.FunnelDTO;
import org.apache.nifi.web.api.dto.LabelDTO;
@@ -58,6 +60,7 @@ import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
+import org.apache.nifi.web.api.dto.ReportingTaskDTO;
import org.apache.nifi.web.api.dto.TemplateDTO;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
@@ -250,6 +253,22 @@ public final class FingerprintFactory {
// root group
final Element rootGroupElem = (Element) DomUtils.getChildNodesByTagName(flowControllerElem, "rootGroup").item(0);
addProcessGroupFingerprint(builder, rootGroupElem, controller);
+
+ final Element controllerServicesElem = DomUtils.getChild(flowControllerElem, "controllerServices");
+ if ( controllerServicesElem != null ) {
+ for ( final Element serviceElem : DomUtils.getChildElementsByTagName(controllerServicesElem, "controllerService") ) {
+ addControllerServiceFingerprint(builder, serviceElem);
+ }
+ }
+
+ final Element reportingTasksElem = DomUtils.getChild(flowControllerElem, "reportingTasks");
+ if ( reportingTasksElem != null ) {
+ for ( final Element taskElem : DomUtils.getChildElementsByTagName(reportingTasksElem, "reportingTask") ) {
+ addReportingTaskFingerprint(builder, taskElem);
+ }
+ }
+
+
return builder;
}
@@ -832,6 +851,68 @@ public final class FingerprintFactory {
builder.append(funnel.getId());
return builder;
}
+
+ private void addControllerServiceFingerprint(final StringBuilder builder, final Element controllerServiceElem) {
+ final ControllerServiceDTO dto = FlowFromDOMFactory.getControllerService(controllerServiceElem, encryptor);
+ addControllerServiceFingerprint(builder, dto);
+ }
+
+ private void addControllerServiceFingerprint(final StringBuilder builder, final ControllerServiceDTO dto) {
+ builder.append(dto.getId());
+ builder.append(dto.getType());
+ builder.append(dto.getName());
+ builder.append(dto.getComment());
+ builder.append(dto.getAvailability());
+ builder.append(dto.getAnnotationData());
+
+ final Map<String, String> properties = dto.getProperties();
+ if (properties == null) {
+ builder.append("NO_PROPERTIES");
+ } else {
+ final SortedMap<String, String> sortedProps = new TreeMap<>(properties);
+ for (final Map.Entry<String, String> entry : sortedProps.entrySet()) {
+ final String propName = entry.getKey();
+ final String propValue = entry.getValue();
+ if (propValue == null) {
+ continue;
+ }
+
+ builder.append(propName).append("=").append(propValue);
+ }
+ }
+ }
+
+ private void addReportingTaskFingerprint(final StringBuilder builder, final Element element) {
+ final ReportingTaskDTO dto = FlowFromDOMFactory.getReportingTask(element, encryptor);
+ addReportingTaskFingerprint(builder, dto);
+ }
+
+ private void addReportingTaskFingerprint(final StringBuilder builder, final ReportingTaskDTO dto) {
+ builder.append(dto.getId());
+ builder.append(dto.getType());
+ builder.append(dto.getName());
+ builder.append(dto.getComment());
+ builder.append(dto.getSchedulingPeriod());
+ builder.append(dto.getSchedulingStrategy());
+ builder.append(dto.getAvailability());
+ builder.append(dto.getAnnotationData());
+
+ final Map<String, String> properties = dto.getProperties();
+ if (properties == null) {
+ builder.append("NO_PROPERTIES");
+ } else {
+ final SortedMap<String, String> sortedProps = new TreeMap<>(properties);
+ for (final Map.Entry<String, String> entry : sortedProps.entrySet()) {
+ final String propName = entry.getKey();
+ final String propValue = entry.getValue();
+ if (propValue == null) {
+ continue;
+ }
+
+ builder.append(propName).append("=").append(propValue);
+ }
+ }
+ }
private Comparator<Element> getIdsComparator() {
return new Comparator<Element>() {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/2df4500c/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index 5ed25e6..40d3b1b 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@ -404,9 +404,13 @@ public class ControllerFacade implements ControllerServiceProvider {
return flowController.getControllerService(serviceIdentifier);
}
- @Override
public ControllerServiceNode createControllerService(final String type, final boolean firstTimeAdded) {
- return flowController.createControllerService(type, firstTimeAdded);
+ return flowController.createControllerService(type, firstTimeAdded);
+ }
+
+ @Override
+ public ControllerServiceNode createControllerService(final String type, final String id, final boolean firstTimeAdded) {
+ return flowController.createControllerService(type, id, firstTimeAdded);
}
public void removeControllerService(ControllerServiceNode serviceNode) {
[2/5] incubator-nifi git commit: NIFI-250: Allow Controller Services
to be created without id and properties;
allow controller service lookup to provide name of controller service given an
id
Posted by ma...@apache.org.
NIFI-250: Allow Controller Services to be created without id and properties; allow controller service lookup to provide name of controller service given an id
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/7a3d208f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/7a3d208f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/7a3d208f
Branch: refs/heads/NIFI-250
Commit: 7a3d208f031122ab13349ace8aeafac9fe5c54ac
Parents: 80f02e4
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jan 26 11:42:01 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jan 26 11:42:01 2015 -0500
----------------------------------------------------------------------
.../controller/ControllerServiceLookup.java | 7 ++++
...kControllerServiceInitializationContext.java | 5 +++
.../nifi/util/MockControllerServiceLookup.java | 6 ++++
.../MockProcessorInitializationContext.java | 5 +++
.../apache/nifi/util/MockValidationContext.java | 6 ++++
.../cluster/manager/impl/WebClusterManager.java | 10 ++++--
.../service/ControllerServiceProvider.java | 7 ++--
.../service/ControllerServiceReference.java | 7 ++--
.../apache/nifi/controller/FlowController.java | 38 +++++++++++++++++---
.../reporting/StandardReportingContext.java | 6 ++++
.../StandardReportingInitializationContext.java | 5 +++
...dControllerServiceInitializationContext.java | 5 +++
.../service/StandardControllerServiceNode.java | 2 +-
.../StandardControllerServiceProvider.java | 31 +++++++---------
.../StandardControllerServiceReference.java | 15 ++++----
.../nifi/processor/StandardProcessContext.java | 5 +++
.../processor/TestStandardPropertyValue.java | 6 +++-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 5 +--
.../nifi/web/controller/ControllerFacade.java | 11 ++++--
19 files changed, 133 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java b/nifi/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java
index 77b8e62..25167ad 100644
--- a/nifi/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java
+++ b/nifi/nifi-api/src/main/java/org/apache/nifi/controller/ControllerServiceLookup.java
@@ -63,4 +63,11 @@ public interface ControllerServiceLookup {
*/
Set<String> getControllerServiceIdentifiers(Class<? extends ControllerService> serviceType) throws IllegalArgumentException;
+ /**
+ * Returns the name of the Controller service with the given identifier. If no service can be
+ * found with this identifier, returns {@code null}.
+ * @param serviceIdentifier
+ * @return
+ */
+ String getControllerServiceName(String serviceIdentifier);
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
index 86624ae..1d74029 100644
--- a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
+++ b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceInitializationContext.java
@@ -33,6 +33,11 @@ public class MockControllerServiceInitializationContext extends MockControllerSe
public String getIdentifier() {
return identifier;
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return null;
+ }
@Override
public ControllerServiceLookup getControllerServiceLookup() {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
index ed0d650..82540d0 100644
--- a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
+++ b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockControllerServiceLookup.java
@@ -73,4 +73,10 @@ public abstract class MockControllerServiceLookup implements ControllerServiceLo
}
return ids;
}
+
+ @Override
+ public String getControllerServiceName(String serviceIdentifier) {
+ final ControllerServiceConfiguration status = controllerServiceMap.get(serviceIdentifier);
+ return status == null ? null : serviceIdentifier;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
index f49a6c5..582ffb1 100644
--- a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
+++ b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
@@ -63,6 +63,11 @@ public class MockProcessorInitializationContext implements ProcessorInitializati
}
@Override
+ public String getControllerServiceName(String serviceIdentifier) {
+ return context.getControllerServiceName(serviceIdentifier);
+ }
+
+ @Override
public boolean isControllerServiceEnabled(String serviceIdentifier) {
return context.isControllerServiceEnabled(serviceIdentifier);
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
index fd0b6cd..34fd7de 100644
--- a/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
+++ b/nifi/nifi-mock/src/main/java/org/apache/nifi/util/MockValidationContext.java
@@ -90,4 +90,10 @@ public class MockValidationContext implements ValidationContext, ControllerServi
public boolean isControllerServiceEnabled(final ControllerService service) {
return context.isControllerServiceEnabled(service);
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ final ControllerServiceConfiguration configuration = context.getConfiguration(serviceIdentifier);
+ return configuration == null ? null : serviceIdentifier;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/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 511bb7d..17a6452 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
@@ -191,7 +191,6 @@ import org.apache.nifi.web.api.entity.ProvenanceEventEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupsEntity;
import org.apache.nifi.web.util.WebUtils;
-
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1299,8 +1298,8 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
* @return
*/
@Override
- public ControllerServiceNode createControllerService(String type, String id, Map<String, String> properties) {
- return controllerServiceProvider.createControllerService(type, id, properties);
+ public ControllerServiceNode createControllerService(String type) {
+ return controllerServiceProvider.createControllerService(type);
}
@Override
@@ -1323,6 +1322,11 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return controllerServiceProvider.isControllerServiceEnabled(serviceIdentifier);
}
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return controllerServiceProvider.getControllerServiceName(serviceIdentifier);
+ }
+
/**
* Handle a bulletins message.
*
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
index 35a255d..ba00799 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
@@ -16,8 +16,6 @@
*/
package org.apache.nifi.controller.service;
-import java.util.Map;
-
import org.apache.nifi.controller.ControllerServiceLookup;
/**
@@ -26,15 +24,14 @@ import org.apache.nifi.controller.ControllerServiceLookup;
public interface ControllerServiceProvider extends ControllerServiceLookup {
/**
- * Gets the controller service for the specified identifier. Returns null if
- * the identifier does not match a known service.
+ * Creates a new Controller Service of the specified type
*
* @param type
* @param id
* @param properties
* @return
*/
- ControllerServiceNode createControllerService(String type, String id, Map<String, String> properties);
+ ControllerServiceNode createControllerService(String type);
/**
* Gets the controller service node for the specified identifier. Returns
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceReference.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceReference.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceReference.java
index 5cb676f..67ffb6c 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceReference.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceReference.java
@@ -41,10 +41,11 @@ public interface ControllerServiceReference {
Set<ConfiguredComponent> getReferencingComponents();
/**
- * Returns a {@link Set} of all Processors and Reporting Tasks that are
- * referencing the Controller Service and are running, in addition to all
+ * Returns a {@link Set} of all Processors, Reporting Tasks, and Controller Services that are
+ * referencing the Controller Service and are running (in the case of Processors and Reporting Tasks)
+ * or enabled (in the case of Controller Services)
*
* @return
*/
- Set<ConfiguredComponent> getRunningReferences();
+ Set<ConfiguredComponent> getActiveReferences();
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/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 346e801..ae0ef97 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
@@ -48,6 +48,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.net.ssl.SSLContext;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.admin.service.UserService;
import org.apache.nifi.cluster.BulletinsPayload;
import org.apache.nifi.cluster.HeartbeatPayload;
@@ -100,6 +101,7 @@ import org.apache.nifi.controller.scheduling.StandardProcessScheduler;
import org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.controller.service.ControllerServiceReference;
import org.apache.nifi.controller.service.StandardControllerServiceProvider;
import org.apache.nifi.controller.status.ConnectionStatus;
import org.apache.nifi.controller.status.PortStatus;
@@ -126,7 +128,6 @@ import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroupPortDescriptor;
import org.apache.nifi.groups.StandardProcessGroup;
-import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.logging.LogRepository;
import org.apache.nifi.logging.LogRepositoryFactory;
@@ -163,6 +164,7 @@ import org.apache.nifi.reporting.EventAccess;
import org.apache.nifi.reporting.ReportingTask;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.FormatUtils;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.ReflectionUtils;
@@ -182,7 +184,6 @@ import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
import org.apache.nifi.web.api.dto.TemplateDTO;
import org.apache.nifi.web.api.dto.status.StatusHistoryDTO;
-import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -2522,9 +2523,32 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
return reportingTasks.values();
}
+ /**
+ * Recursively stops all Processors and Reporting Tasks that are referencing the given Controller Service,
+ * as well as disabling any Controller Service that references this Controller Service (and stops
+ * all Reporting Task or Controller Service that is referencing it, and so on).
+ * @param serviceNode
+ */
+ public void deactiveReferencingComponents(final ControllerServiceNode serviceNode) {
+ final ControllerServiceReference reference = serviceNode.getReferences();
+
+ final Set<ConfiguredComponent> components = reference.getActiveReferences();
+ for (final ConfiguredComponent component : components) {
+ if ( component instanceof ControllerServiceNode ) {
+ deactiveReferencingComponents((ControllerServiceNode) component);
+ // TODO: DISABLE CONTROLLER SERVICE!
+ } else if ( component instanceof ReportingTaskNode ) {
+ stopReportingTask((ReportingTaskNode) component);
+ } else if ( component instanceof ProcessorNode ) {
+ final ProcessorNode procNode = (ProcessorNode) component;
+ stopProcessor(procNode.getProcessGroup().getIdentifier(), procNode.getIdentifier());
+ }
+ }
+ }
+
@Override
- public ControllerServiceNode createControllerService(final String type, final String id, final Map<String, String> properties) {
- return controllerServiceProvider.createControllerService(type, id.intern(), properties);
+ public ControllerServiceNode createControllerService(final String type) {
+ return controllerServiceProvider.createControllerService(type);
}
@Override
@@ -2547,6 +2571,12 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
return controllerServiceProvider.isControllerServiceEnabled(serviceIdentifier);
}
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return controllerServiceProvider.getControllerServiceName(serviceIdentifier);
+ }
+
+
//
// Counters
//
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
index ed48e20..2b49d76 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingContext.java
@@ -129,4 +129,10 @@ public class StandardReportingContext implements ReportingContext, ControllerSer
public ControllerServiceLookup getControllerServiceLookup() {
return this;
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return serviceProvider.getControllerServiceName(serviceIdentifier);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
index d576f9c..6330373 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingInitializationContext.java
@@ -93,4 +93,9 @@ public class StandardReportingInitializationContext implements ReportingInitiali
public ControllerServiceLookup getControllerServiceLookup() {
return this;
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return serviceProvider.getControllerServiceName(serviceIdentifier);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
index 8b5f27f..ed7a152 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceInitializationContext.java
@@ -61,4 +61,9 @@ public class StandardControllerServiceInitializationContext implements Controlle
public boolean isControllerServiceEnabled(final ControllerService service) {
return serviceProvider.isControllerServiceEnabled(service);
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return serviceProvider.getControllerServiceName(serviceIdentifier);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index 455eac1..e3e6e6b 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -62,7 +62,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
if (disabled) {
// do not allow a Controller Service to be disabled if it's currently being used.
- final Set<ConfiguredComponent> runningRefs = getReferences().getRunningReferences();
+ final Set<ConfiguredComponent> runningRefs = getReferences().getActiveReferences();
if (!runningRefs.isEmpty()) {
throw new IllegalStateException("Cannot disable Controller Service because it is referenced (either directly or indirectly) by " + runningRefs.size() + " different components that are currently running");
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index fc07ce1..66cc541 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -28,19 +28,16 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ValidationContextFactory;
-import org.apache.nifi.controller.annotation.OnConfigured;
-import org.apache.nifi.controller.exception.ControllerServiceAlreadyExistsException;
import org.apache.nifi.controller.exception.ControllerServiceNotFoundException;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.StandardValidationContextFactory;
import org.apache.nifi.util.ObjectHolder;
-import org.apache.nifi.util.ReflectionUtils;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,14 +90,12 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
}
@Override
- public ControllerServiceNode createControllerService(final String type, final String id, final Map<String, String> properties) {
- if (type == null || id == null) {
+ public ControllerServiceNode createControllerService(final String type) {
+ if (type == null) {
throw new NullPointerException();
}
- if (controllerServices.containsKey(id)) {
- throw new ControllerServiceAlreadyExistsException(id);
- }
-
+
+ final String id = UUID.randomUUID().toString();
final ClassLoader currentContextClassLoader = Thread.currentThread().getContextClassLoader();
try {
final ClassLoader cl = ExtensionManager.getClassLoader(type);
@@ -133,7 +128,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
};
final ControllerService proxiedService = (ControllerService) Proxy.newProxyInstance(cl, getInterfaces(controllerServiceClass), invocationHandler);
- logger.info("Loaded service {} as configured.", type);
+ logger.info("Create Controller Service of type {} with identifier {}", type, id);
originalService.initialize(new StandardControllerServiceInitializationContext(id, this));
@@ -141,13 +136,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
final ControllerServiceNode serviceNode = new StandardControllerServiceNode(proxiedService, id, validationContextFactory, this);
serviceNodeHolder.set(serviceNode);
- serviceNode.setAnnotationData(null);
- serviceNode.setName(id);
- for (final Map.Entry<String, String> entry : properties.entrySet()) {
- serviceNode.setProperty(entry.getKey(), entry.getValue());
- }
- final StandardConfigurationContext configurationContext = new StandardConfigurationContext(serviceNode, this);
- ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnConfigured.class, originalService, configurationContext);
+ serviceNode.setName(rawClass.getSimpleName());
this.controllerServices.put(id, serviceNode);
return serviceNode;
@@ -193,4 +182,10 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
return identifiers;
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ final ControllerServiceNode node = getControllerServiceNode(serviceIdentifier);
+ return node == null ? null : node.getName();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceReference.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceReference.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceReference.java
index a1c4984..a8468ff 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceReference.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceReference.java
@@ -58,23 +58,24 @@ public class StandardControllerServiceReference implements ControllerServiceRefe
}
@Override
- public Set<ConfiguredComponent> getRunningReferences() {
- final Set<ConfiguredComponent> runningReferences = new HashSet<>();
+ public Set<ConfiguredComponent> getActiveReferences() {
+ final Set<ConfiguredComponent> activeReferences = new HashSet<>();
final Set<ControllerServiceNode> serviceNodes = new HashSet<>();
for (final ConfiguredComponent component : components) {
if (component instanceof ControllerServiceNode) {
serviceNodes.add((ControllerServiceNode) component);
+ activeReferences.add(component);
} else if (isRunning(component)) {
- runningReferences.add(component);
+ activeReferences.add(component);
}
}
- runningReferences.addAll(getRunningIndirectReferences(serviceNodes));
- return runningReferences;
+ activeReferences.addAll(getActiveIndirectReferences(serviceNodes));
+ return activeReferences;
}
- private Set<ConfiguredComponent> getRunningIndirectReferences(final Set<ControllerServiceNode> referencingServices) {
+ private Set<ConfiguredComponent> getActiveIndirectReferences(final Set<ControllerServiceNode> referencingServices) {
if (referencingServices.isEmpty()) {
return Collections.emptySet();
}
@@ -92,7 +93,7 @@ public class StandardControllerServiceReference implements ControllerServiceRefe
}
}
- references.addAll(getRunningIndirectReferences(serviceNodes));
+ references.addAll(getActiveIndirectReferences(serviceNodes));
}
return references;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
index cd0d31c..8d6b710 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
@@ -170,4 +170,9 @@ public class StandardProcessContext implements ProcessContext, ControllerService
return set;
}
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return controllerServiceProvider.getControllerServiceName(serviceIdentifier);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
index 7390098..2c85655 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/processor/TestStandardPropertyValue.java
@@ -17,6 +17,7 @@
package org.apache.nifi.processor;
import org.apache.nifi.processor.StandardPropertyValue;
+
import static org.junit.Assert.assertEquals;
import java.util.Calendar;
@@ -29,7 +30,6 @@ import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.repository.StandardFlowFileRecord;
import org.apache.nifi.flowfile.FlowFile;
-
import org.junit.Test;
public class TestStandardPropertyValue {
@@ -163,5 +163,9 @@ public class TestStandardPropertyValue {
return true;
}
+ @Override
+ public String getControllerServiceName(String serviceIdentifier) {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 7d44624..f2d6325 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -1762,10 +1762,7 @@ public final class DtoFactory {
} else {
final Set<AllowableValueDTO> allowableValues = new LinkedHashSet<>();
for (final String serviceIdentifier : controllerServiceLookup.getControllerServiceIdentifiers(serviceDefinition)) {
- String displayName = serviceIdentifier;
-
- // TODO: attempt to get the controller service name
- final ControllerService controllerService = controllerServiceLookup.getControllerService(serviceIdentifier);
+ final String displayName = controllerServiceLookup.getControllerServiceName(serviceIdentifier);
final AllowableValueDTO allowableValue = new AllowableValueDTO();
allowableValue.setDisplayName(displayName);
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7a3d208f/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index 4a00d02..27c60bb 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@ -115,7 +115,6 @@ import org.apache.nifi.admin.service.UserService;
import org.apache.nifi.authorization.DownloadAuthorization;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.reporting.ReportingTask;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.access.AccessDeniedException;
@@ -406,8 +405,8 @@ public class ControllerFacade implements ControllerServiceProvider {
}
@Override
- public ControllerServiceNode createControllerService(String type, String id, Map<String, String> properties) {
- return flowController.createControllerService(type, id, properties);
+ public ControllerServiceNode createControllerService(String type) {
+ return flowController.createControllerService(type);
}
@Override
@@ -429,6 +428,12 @@ public class ControllerFacade implements ControllerServiceProvider {
public boolean isControllerServiceEnabled(final String serviceIdentifier) {
return flowController.isControllerServiceEnabled(serviceIdentifier);
}
+
+ @Override
+ public String getControllerServiceName(final String serviceIdentifier) {
+ return flowController.getControllerServiceName(serviceIdentifier);
+ }
+
/**
* Gets the status of this controller.
[4/5] incubator-nifi git commit: NIFI-250: Serialize and deserialize
controller services and reporting tasks in the flow.xml
Posted by ma...@apache.org.
NIFI-250: Serialize and deserialize controller services and reporting tasks in the flow.xml
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/86d15f9e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/86d15f9e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/86d15f9e
Branch: refs/heads/NIFI-250
Commit: 86d15f9e1c3b28a9ae587a1cb0c73cd2b6e8b51c
Parents: ff43b03
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jan 26 13:55:55 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jan 26 13:55:55 2015 -0500
----------------------------------------------------------------------
.../nifi/web/api/dto/ControllerServiceDTO.java | 16 +-
.../nifi/web/api/dto/ProcessorConfigDTO.java | 6 +-
.../nifi/web/api/dto/ReportingTaskDTO.java | 166 +++++++++++++++++++
.../cluster/manager/impl/WebClusterManager.java | 4 +
.../nifi/controller/ReportingTaskNode.java | 4 +
.../service/ControllerServiceNode.java | 3 +
.../service/ControllerServiceProvider.java | 8 +
.../apache/nifi/controller/FlowController.java | 31 +++-
.../nifi/controller/FlowFromDOMFactory.java | 59 +++++--
.../nifi/controller/StandardFlowSerializer.java | 62 ++++++-
.../controller/StandardFlowSynchronizer.java | 61 ++++++-
.../reporting/AbstractReportingTaskNode.java | 16 +-
.../service/ControllerServiceLoader.java | 3 +-
.../service/StandardControllerServiceNode.java | 21 +++
.../StandardControllerServiceProvider.java | 10 +-
.../StandardXMLFlowConfigurationDAO.java | 10 +-
.../nifi/web/controller/ControllerFacade.java | 4 +
17 files changed, 439 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ControllerServiceDTO.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ControllerServiceDTO.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ControllerServiceDTO.java
index 7662594..94ac55a 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ControllerServiceDTO.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ControllerServiceDTO.java
@@ -20,13 +20,14 @@ import java.util.Map;
import javax.xml.bind.annotation.XmlType;
/**
- * A connection between two connectable components.
+ * A Controller Service that can be shared by other components
*/
@XmlType(name = "controllerService")
public class ControllerServiceDTO extends NiFiComponentDTO {
private String name;
private String type;
+ private String comment;
private String availability;
private Boolean enabled;
@@ -60,8 +61,21 @@ public class ControllerServiceDTO extends NiFiComponentDTO {
public void setType(String type) {
this.type = type;
}
+
/**
+ * The comment for the Controller Service
+ * @return
+ */
+ public String getComment() {
+ return comment;
+ }
+
+ public void setComment(String comment) {
+ this.comment = comment;
+ }
+
+ /**
* Where this service is available. Possible values are CLUSTER_MANAGER_ONLY, NODE_ONLY, BOTH.
*
* @return
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessorConfigDTO.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessorConfigDTO.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessorConfigDTO.java
index 5978041..d1a44b8 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessorConfigDTO.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessorConfigDTO.java
@@ -54,10 +54,10 @@ public class ProcessorConfigDTO {
}
/**
- * The amount of time that should elapse between task executions. This will
- * not affect currently scheduled tasks.
+ * The frequency with which to schedule the processor. The format of the value will
+ * depend on the value of {@link #getSchedulingStrategy()}.
*
- * @return The scheduling period in seconds
+ * @return The scheduling period
*/
public String getSchedulingPeriod() {
return schedulingPeriod;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ReportingTaskDTO.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ReportingTaskDTO.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ReportingTaskDTO.java
new file mode 100644
index 0000000..a04e9bb
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ReportingTaskDTO.java
@@ -0,0 +1,166 @@
+/*
+ * 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.web.api.dto;
+
+import java.util.Map;
+
+import javax.xml.bind.annotation.XmlType;
+
+/**
+ * Component that is capable of reporting internal NiFi state to an external service
+ */
+@XmlType(name = "reportingTask")
+public class ReportingTaskDTO extends NiFiComponentDTO {
+ private String name;
+ private String comment;
+ private String type;
+ private String schedulingPeriod;
+ private String scheduledState;
+ private String schedulingStrategy;
+ private String availability;
+
+ private Map<String, String> properties;
+ private Map<String, PropertyDescriptorDTO> descriptors;
+
+ private String annotationData;
+
+ /**
+ * The user-defined name of the reporting task
+ * @return
+ */
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ /**
+ * The user-defined comment for the reporting task
+ * @return
+ */
+ public String getComment() {
+ return comment;
+ }
+
+ public void setComment(String comment) {
+ this.comment = comment;
+ }
+
+ /**
+ * The type of reporting task
+ * @return
+ */
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ /**
+ * The frequency with which to schedule the reporting task. The format of the value will
+ * depend on the value of {@link #getSchedulingStrategy()}.
+ *
+ * @return The scheduling period
+ */
+ public String getSchedulingPeriod() {
+ return schedulingPeriod;
+ }
+
+ public void setSchedulingPeriod(String schedulingPeriod) {
+ this.schedulingPeriod = schedulingPeriod;
+ }
+
+ /**
+ * The current scheduling state of the reporting task
+ * @return
+ */
+ public String getScheduledState() {
+ return scheduledState;
+ }
+
+ public void setScheduledState(String scheduledState) {
+ this.scheduledState = scheduledState;
+ }
+
+ /**
+ * The scheduling strategy that determines how the {@link #getSchedulingPeriod()} value should
+ * be interpreted
+ *
+ * @return
+ */
+ public String getSchedulingStrategy() {
+ return schedulingStrategy;
+ }
+
+ public void setSchedulingStrategy(String schedulingStrategy) {
+ this.schedulingStrategy = schedulingStrategy;
+ }
+
+ /**
+ * Where this service is available. Possible values are CLUSTER_MANAGER_ONLY, NODE_ONLY, BOTH.
+ *
+ * @return
+ */
+ public String getAvailability() {
+ return availability;
+ }
+
+ public void setAvailability(String availability) {
+ this.availability = availability;
+ }
+
+ /**
+ * The reporting task's properties
+ * @return
+ */
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
+ /**
+ * Map of property name to descriptor
+ * @return
+ */
+ public Map<String, PropertyDescriptorDTO> getDescriptors() {
+ return descriptors;
+ }
+
+ public void setDescriptors(Map<String, PropertyDescriptorDTO> descriptors) {
+ this.descriptors = descriptors;
+ }
+
+ /**
+ * The currently configured annotation data for the reporting task
+ * @return
+ */
+ public String getAnnotationData() {
+ return annotationData;
+ }
+
+ public void setAnnotationData(String annotationData) {
+ this.annotationData = annotationData;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/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 5f1c460..54c2b55 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
@@ -1342,6 +1342,10 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
controllerServiceProvider.disableControllerService(serviceNode);
}
+ @Override
+ public Set<ControllerServiceNode> getAllControllerServices() {
+ return controllerServiceProvider.getAllControllerServices();
+ }
/**
* Handle a bulletins message.
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReportingTaskNode.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReportingTaskNode.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReportingTaskNode.java
index fa48cb3..cf81352 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReportingTaskNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReportingTaskNode.java
@@ -68,6 +68,10 @@ public interface ReportingTaskNode extends ConfiguredComponent {
void setScheduledState(ScheduledState state);
+ String getComment();
+
+ void setComment(String comment);
+
void verifyCanStart();
void verifyCanStop();
void verifyCanDisable();
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
index 66bad39..32b9d9e 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
@@ -40,6 +40,9 @@ public interface ControllerServiceNode extends ConfiguredComponent {
void removeReference(ConfiguredComponent referringComponent);
+ void setComment(String comment);
+ String getComment();
+
void verifyCanEnable();
void verifyCanDisable();
void verifyCanDelete();
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
index 54a506e..60ff6c9 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceProvider.java
@@ -16,6 +16,8 @@
*/
package org.apache.nifi.controller.service;
+import java.util.Set;
+
import org.apache.nifi.annotation.lifecycle.OnAdded;
import org.apache.nifi.controller.ControllerServiceLookup;
@@ -65,4 +67,10 @@ public interface ControllerServiceProvider extends ControllerServiceLookup {
* @param serviceNode
*/
void disableControllerService(ControllerServiceNode serviceNode);
+
+ /**
+ * Returns a Set of all Controller Services that exist for this service provider.
+ * @return
+ */
+ Set<ControllerServiceNode> getAllControllerServices();
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/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 38c8d42..92975ad 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
@@ -2464,16 +2464,16 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
lookupGroup(groupId).stopProcessing();
}
- public ReportingTaskNode createReportingTask(final String type, String id) throws ReportingTaskInstantiationException {
- return createReportingTask(type, id, true);
+ public ReportingTaskNode createReportingTask(final String type) throws ReportingTaskInstantiationException {
+ return createReportingTask(type, true);
}
- public ReportingTaskNode createReportingTask(final String type, String id, final boolean firstTimeAdded) throws ReportingTaskInstantiationException {
+ public ReportingTaskNode createReportingTask(final String type, final boolean firstTimeAdded) throws ReportingTaskInstantiationException {
if (type == null) {
throw new NullPointerException();
}
-
- id = requireNonNull(id).intern();
+
+ final String id = UUID.randomUUID().toString();
ReportingTask task = null;
final ClassLoader ctxClassLoader = Thread.currentThread().getContextClassLoader();
@@ -2563,18 +2563,28 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
* @param serviceNode
*/
public void deactiveReferencingComponents(final ControllerServiceNode serviceNode) {
+ // TODO: Should stop all Processors and Reporting Tasks, and then wait for them to
+ // finish, rather than stopping & waiting serially.
final ControllerServiceReference reference = serviceNode.getReferences();
final Set<ConfiguredComponent> components = reference.getActiveReferences();
for (final ConfiguredComponent component : components) {
if ( component instanceof ControllerServiceNode ) {
deactiveReferencingComponents((ControllerServiceNode) component);
- // TODO: DISABLE CONTROLLER SERVICE!
+
+ if (isControllerServiceEnabled(serviceNode.getIdentifier())) {
+ disableControllerService(serviceNode);
+ }
} else if ( component instanceof ReportingTaskNode ) {
- stopReportingTask((ReportingTaskNode) component);
+ final ReportingTaskNode taskNode = (ReportingTaskNode) component;
+ if (taskNode.isRunning()) {
+ stopReportingTask((ReportingTaskNode) component);
+ }
} else if ( component instanceof ProcessorNode ) {
final ProcessorNode procNode = (ProcessorNode) component;
- stopProcessor(procNode.getProcessGroup().getIdentifier(), procNode.getIdentifier());
+ if ( procNode.isRunning() ) {
+ stopProcessor(procNode.getProcessGroup().getIdentifier(), procNode.getIdentifier());
+ }
}
}
}
@@ -2637,6 +2647,11 @@ public class FlowController implements EventAccess, ControllerServiceProvider, H
controllerServiceProvider.removeControllerService(serviceNode);
}
+ @Override
+ public Set<ControllerServiceNode> getAllControllerServices() {
+ return controllerServiceProvider.getAllControllerServices();
+ }
+
//
// Counters
//
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowFromDOMFactory.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowFromDOMFactory.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowFromDOMFactory.java
index c67181a..d0dc83e 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowFromDOMFactory.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowFromDOMFactory.java
@@ -33,6 +33,7 @@ import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.DomUtils;
import org.apache.nifi.web.api.dto.ConnectableDTO;
import org.apache.nifi.web.api.dto.ConnectionDTO;
+import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.FunnelDTO;
import org.apache.nifi.web.api.dto.LabelDTO;
@@ -42,7 +43,7 @@ import org.apache.nifi.web.api.dto.ProcessGroupDTO;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
-
+import org.apache.nifi.web.api.dto.ReportingTaskDTO;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
@@ -77,6 +78,40 @@ public class FlowFromDOMFactory {
return styles;
}
+
+ public static ControllerServiceDTO getControllerService(final Element element, final StringEncryptor encryptor) {
+ final ControllerServiceDTO dto = new ControllerServiceDTO();
+
+ dto.setId(getString(element, "id"));
+ dto.setName(getString(element, "name"));
+ dto.setComment(getString(element, "comment"));
+ dto.setType(getString(element, "class"));
+ dto.setEnabled(getBoolean(element, "enabled"));
+ dto.setAvailability(getString(element, "availability"));
+
+ dto.setProperties(getProperties(element, encryptor));
+ dto.setAnnotationData(getString(element, "annotationData"));
+
+ return dto;
+ }
+
+ public static ReportingTaskDTO getReportingTask(final Element element, final StringEncryptor encryptor) {
+ final ReportingTaskDTO dto = new ReportingTaskDTO();
+
+ dto.setId(getString(element, "id"));
+ dto.setName(getString(element, "name"));
+ dto.setComment(getString(element, "comment"));
+ dto.setType(getString(element, "class"));
+ dto.setSchedulingPeriod(getString(element, "schedulingPeriod"));
+ dto.setScheduledState(getString(element, "scheduledState"));
+ dto.setSchedulingStrategy(getString(element, "schedulingStrategy"));
+ dto.setAvailability(getString(element, "availability"));
+
+ dto.setProperties(getProperties(element, encryptor));
+ dto.setAnnotationData(getString(element, "annotationData"));
+
+ return dto;
+ }
public static ProcessGroupDTO getProcessGroup(final String parentId, final Element element, final StringEncryptor encryptor) {
final ProcessGroupDTO dto = new ProcessGroupDTO();
@@ -310,7 +345,6 @@ public class FlowFromDOMFactory {
final ProcessorConfigDTO configDto = new ProcessorConfigDTO();
dto.setConfig(configDto);
configDto.setComments(getString(element, "comment"));
- configDto.setAnnotationData(getString(element, "annotationData"));
configDto.setConcurrentlySchedulableTaskCount(getInt(element, "maxConcurrentTasks"));
final String schedulingPeriod = getString(element, "schedulingPeriod");
configDto.setSchedulingPeriod(schedulingPeriod);
@@ -334,14 +368,8 @@ public class FlowFromDOMFactory {
configDto.setRunDurationMillis(TimeUnit.NANOSECONDS.toMillis(runDurationNanos));
}
- final LinkedHashMap<String, String> properties = new LinkedHashMap<>();
- final List<Element> propertyNodeList = getChildrenByTagName(element, "property");
- for (final Element propertyElement : propertyNodeList) {
- final String name = getString(propertyElement, "name");
- final String value = decrypt(getString(propertyElement, "value"), encryptor);
- properties.put(name, value);
- }
- configDto.setProperties(properties);
+ configDto.setProperties(getProperties(element, encryptor));
+ configDto.setAnnotationData(getString(element, "annotationData"));
final Set<String> autoTerminatedRelationships = new HashSet<>();
final List<Element> autoTerminateList = getChildrenByTagName(element, "autoTerminatedRelationship");
@@ -353,6 +381,17 @@ public class FlowFromDOMFactory {
return dto;
}
+ private static LinkedHashMap<String, String> getProperties(final Element element, final StringEncryptor encryptor) {
+ final LinkedHashMap<String, String> properties = new LinkedHashMap<>();
+ final List<Element> propertyNodeList = getChildrenByTagName(element, "property");
+ for (final Element propertyElement : propertyNodeList) {
+ final String name = getString(propertyElement, "name");
+ final String value = decrypt(getString(propertyElement, "value"), encryptor);
+ properties.put(name, value);
+ }
+ return properties;
+ }
+
private static String getString(final Element element, final String childElementName) {
final List<Element> nodeList = getChildrenByTagName(element, childElementName);
if (nodeList == null || nodeList.isEmpty()) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
index e08a94d..218813a 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
@@ -40,6 +40,7 @@ import org.apache.nifi.connectable.Port;
import org.apache.nifi.connectable.Position;
import org.apache.nifi.connectable.Size;
import org.apache.nifi.controller.label.Label;
+import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.groups.ProcessGroup;
@@ -47,7 +48,6 @@ import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.remote.RootGroupPort;
-
import org.w3c.dom.DOMException;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
@@ -79,6 +79,18 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(rootNode, "maxTimerDrivenThreadCount", controller.getMaxTimerDrivenThreadCount());
addTextElement(rootNode, "maxEventDrivenThreadCount", controller.getMaxEventDrivenThreadCount());
addProcessGroup(rootNode, controller.getGroup(controller.getRootGroupId()), "rootGroup");
+
+ final Element controllerServicesNode = doc.createElement("controllerServices");
+ rootNode.appendChild(controllerServicesNode);
+ for ( final ControllerServiceNode serviceNode : controller.getAllControllerServices() ) {
+ addControllerService(controllerServicesNode, serviceNode);
+ }
+
+ final Element reportingTasksNode = doc.createElement("reportingTasks");
+ rootNode.appendChild(reportingTasksNode);
+ for ( final ReportingTaskNode taskNode : controller.getReportingTasks() ) {
+ addReportingTask(reportingTasksNode, taskNode);
+ }
final DOMSource domSource = new DOMSource(doc);
final StreamResult streamResult = new StreamResult(new BufferedOutputStream(os));
@@ -300,8 +312,16 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "schedulingStrategy", processor.getSchedulingStrategy().name());
addTextElement(element, "runDurationNanos", processor.getRunDuration(TimeUnit.NANOSECONDS));
- // properties.
- for (final Map.Entry<PropertyDescriptor, String> entry : processor.getProperties().entrySet()) {
+ addConfiguration(element, processor.getProperties(), processor.getAnnotationData());
+
+ for (final Relationship rel : processor.getAutoTerminatedRelationships()) {
+ addTextElement(element, "autoTerminatedRelationship", rel.getName());
+ }
+ }
+
+ private void addConfiguration(final Element element, final Map<PropertyDescriptor, String> properties, final String annotationData) {
+ final Document doc = element.getOwnerDocument();
+ for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) {
final PropertyDescriptor descriptor = entry.getKey();
String value = entry.getValue();
@@ -322,14 +342,9 @@ public class StandardFlowSerializer implements FlowSerializer {
element.appendChild(propElement);
}
- final String annotationData = processor.getAnnotationData();
if (annotationData != null) {
addTextElement(element, "annotationData", annotationData);
}
-
- for (final Relationship rel : processor.getAutoTerminatedRelationships()) {
- addTextElement(element, "autoTerminatedRelationship", rel.getName());
- }
}
private void addConnection(final Element parentElement, final Connection connection) {
@@ -390,6 +405,37 @@ public class StandardFlowSerializer implements FlowSerializer {
parentElement.appendChild(element);
}
+
+ private void addControllerService(final Element element, final ControllerServiceNode serviceNode) {
+ final Element serviceElement = element.getOwnerDocument().createElement("controllerService");
+ addTextElement(serviceElement, "id", serviceNode.getIdentifier());
+ addTextElement(serviceElement, "name", serviceNode.getName());
+ addTextElement(serviceElement, "comment", serviceNode.getComment());
+ addTextElement(serviceElement, "class", serviceNode.getControllerServiceImplementation().getClass().getCanonicalName());
+ addTextElement(serviceElement, "enabled", String.valueOf(!serviceNode.isDisabled()));
+ addTextElement(serviceElement, "availability", serviceNode.getAvailability().toString());
+
+ addConfiguration(serviceElement, serviceNode.getProperties(), serviceNode.getAnnotationData());
+
+ element.appendChild(serviceElement);
+ }
+
+ private void addReportingTask(final Element element, final ReportingTaskNode taskNode) {
+ final Element taskElement = element.getOwnerDocument().createElement("reportingTask");
+ addTextElement(taskElement, "id", taskNode.getIdentifier());
+ addTextElement(taskElement, "name", taskNode.getName());
+ addTextElement(taskElement, "comment", taskNode.getComment());
+ addTextElement(taskElement, "class", taskNode.getReportingTask().getClass().getCanonicalName());
+ addTextElement(taskElement, "schedulingPeriod", taskNode.getSchedulingPeriod());
+ addTextElement(taskElement, "scheduledState", taskNode.getScheduledState().name());
+ addTextElement(taskElement, "schedulingStrategy", taskNode.getSchedulingStrategy().name());
+ addTextElement(taskElement, "availability", taskNode.getAvailability().toString());
+
+ addConfiguration(taskElement, taskNode.getProperties(), taskNode.getAnnotationData());
+
+ element.appendChild(taskElement);
+ }
+
private void addTextElement(final Element element, final String name, final long value) {
addTextElement(element, name, String.valueOf(value));
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
index b60d187..0964a91 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
@@ -40,6 +40,7 @@ import javax.xml.parsers.ParserConfigurationException;
import javax.xml.validation.Schema;
import javax.xml.validation.SchemaFactory;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.cluster.protocol.StandardDataFlow;
import org.apache.nifi.connectable.Connectable;
@@ -51,8 +52,10 @@ import org.apache.nifi.connectable.Position;
import org.apache.nifi.connectable.Size;
import org.apache.nifi.controller.exception.ProcessorInstantiationException;
import org.apache.nifi.controller.label.Label;
+import org.apache.nifi.controller.reporting.ReportingTaskInstantiationException;
+import org.apache.nifi.controller.service.ControllerServiceNode;
+import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.events.BulletinFactory;
-import org.apache.nifi.util.file.FileUtils;
import org.apache.nifi.fingerprint.FingerprintException;
import org.apache.nifi.fingerprint.FingerprintFactory;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
@@ -65,9 +68,12 @@ import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.remote.RootGroupPort;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.util.DomUtils;
import org.apache.nifi.util.NiFiProperties;
+import org.apache.nifi.util.file.FileUtils;
import org.apache.nifi.web.api.dto.ConnectableDTO;
import org.apache.nifi.web.api.dto.ConnectionDTO;
+import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.FunnelDTO;
import org.apache.nifi.web.api.dto.LabelDTO;
@@ -77,9 +83,7 @@ import org.apache.nifi.web.api.dto.ProcessGroupDTO;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
-
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.nifi.encrypt.StringEncryptor;
+import org.apache.nifi.web.api.dto.ReportingTaskDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.Document;
@@ -229,6 +233,22 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
logger.trace("Updating root process group");
updateProcessGroup(controller, /* parent group */ null, rootGroupElement, encryptor);
}
+
+ final Element controllerServicesElement = (Element) DomUtils.getChild(rootElement, "controllerServices");
+ if ( controllerServicesElement != null ) {
+ final List<Element> serviceElements = DomUtils.getChildElementsByTagName(controllerServicesElement, "controllerService");
+ for ( final Element serviceElement : serviceElements ) {
+ addControllerService(controller, serviceElement, encryptor);
+ }
+ }
+
+ final Element reportingTasksElement = (Element) DomUtils.getChild(rootElement, "reportingTasks");
+ if ( reportingTasksElement != null ) {
+ final List<Element> taskElements = DomUtils.getChildElementsByTagName(reportingTasksElement, "reportingTask");
+ for ( final Element taskElement : taskElements ) {
+ addReportingTask(controller, taskElement, encryptor);
+ }
+ }
}
logger.trace("Synching templates");
@@ -313,6 +333,39 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
return baos.toByteArray();
}
+
+ private void addControllerService(final FlowController controller, final Element controllerServiceElement, final StringEncryptor encryptor) {
+ final ControllerServiceDTO dto = FlowFromDOMFactory.getControllerService(controllerServiceElement, encryptor);
+
+ final ControllerServiceNode node = controller.createControllerService(dto.getType(), false);
+ node.setName(dto.getName());
+ node.setAvailability(Availability.valueOf(dto.getAvailability()));
+ node.setComment(dto.getComment());
+ node.setDisabled(dto.getEnabled() != Boolean.TRUE);
+ node.setAnnotationData(dto.getAnnotationData());
+
+ for ( final Map.Entry<String, String> property : dto.getProperties().entrySet() ) {
+ node.setProperty(property.getKey(), property.getValue());
+ }
+ }
+
+ private void addReportingTask(final FlowController controller, final Element reportingTaskElement, final StringEncryptor encryptor) throws ReportingTaskInstantiationException {
+ final ReportingTaskDTO dto = FlowFromDOMFactory.getReportingTask(reportingTaskElement, encryptor);
+
+ final ReportingTaskNode reportingTask = controller.createReportingTask(dto.getType(), false);
+ reportingTask.setName(dto.getName());
+ reportingTask.setComment(dto.getComment());
+ reportingTask.setAvailability(Availability.valueOf(dto.getAvailability()));
+ reportingTask.setScheduldingPeriod(dto.getSchedulingPeriod());
+ reportingTask.setScheduledState(ScheduledState.valueOf(dto.getScheduledState()));
+ reportingTask.setSchedulingStrategy(SchedulingStrategy.valueOf(dto.getSchedulingStrategy()));
+
+ reportingTask.setAnnotationData(dto.getAnnotationData());
+
+ for ( final Map.Entry<String, String> entry : dto.getProperties().entrySet() ) {
+ reportingTask.setProperty(entry.getKey(), entry.getValue());
+ }
+ }
private ProcessGroup updateProcessGroup(final FlowController controller, final ProcessGroup parentGroup, final Element processGroupElement, final StringEncryptor encryptor) throws ProcessorInstantiationException {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
index 7c3734a..014fe55 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/AbstractReportingTaskNode.java
@@ -46,7 +46,8 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
private final AtomicReference<SchedulingStrategy> schedulingStrategy = new AtomicReference<>(SchedulingStrategy.TIMER_DRIVEN);
private final AtomicReference<String> schedulingPeriod = new AtomicReference<>("5 mins");
private final AtomicReference<Availability> availability = new AtomicReference<>(Availability.NODE_ONLY);
-
+
+ private volatile String comment;
private volatile ScheduledState scheduledState = ScheduledState.STOPPED;
public AbstractReportingTaskNode(final ReportingTask reportingTask, final String id,
@@ -142,7 +143,8 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
return removed;
}
- private void onConfigured() {
+ @SuppressWarnings("deprecation")
+ private void onConfigured() {
// We need to invoke any method annotation with the OnConfigured annotation in order to
// maintain backward compatibility. This will be removed when we remove the old, deprecated annotations.
try (final NarCloseable x = NarCloseable.withNarLoader()) {
@@ -158,6 +160,16 @@ public abstract class AbstractReportingTaskNode extends AbstractConfiguredCompon
}
@Override
+ public String getComment() {
+ return comment;
+ }
+
+ @Override
+ public void setComment(final String comment) {
+ this.comment = comment;
+ }
+
+ @Override
public void verifyCanDelete() {
if (isRunning()) {
throw new IllegalStateException("Cannot delete " + reportingTask + " because it is currently running");
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
index db44b5f..7504d13 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/ControllerServiceLoader.java
@@ -122,11 +122,10 @@ public class ControllerServiceLoader {
for (final Element serviceElement : serviceNodes) {
//get properties for the specific controller task - id, name, class,
//and schedulingPeriod must be set
- final String serviceId = DomUtils.getChild(serviceElement, "identifier").getTextContent().trim();
final String serviceClass = DomUtils.getChild(serviceElement, "class").getTextContent().trim();
//set the class to be used for the configured controller task
- final ControllerServiceNode serviceNode = provider.createControllerService(serviceClass, serviceId, false);
+ final ControllerServiceNode serviceNode = provider.createControllerService(serviceClass, false);
//optional task-specific properties
for (final Element optionalProperty : DomUtils.getChildElementsByTagName(serviceElement, "property")) {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index b829bbd..f5776ce 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -49,6 +49,7 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
private final Lock writeLock = rwLock.writeLock();
private final Set<ConfiguredComponent> referencingComponents = new HashSet<>();
+ private String comment;
public StandardControllerServiceNode(final ControllerService proxiedControllerService, final ControllerService implementation, final String id,
final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider) {
@@ -193,4 +194,24 @@ public class StandardControllerServiceNode extends AbstractConfiguredComponent i
throw new IllegalStateException(implementation + " cannot be updated because it is not disabled");
}
}
+
+ @Override
+ public String getComment() {
+ readLock.lock();
+ try {
+ return comment;
+ } finally {
+ readLock.unlock();
+ }
+ }
+
+ @Override
+ public void setComment(final String comment) {
+ writeLock.lock();
+ try {
+ this.comment = comment;
+ } finally {
+ writeLock.unlock();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index b07fc3f..75de4b9 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -30,10 +30,11 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import org.apache.nifi.annotation.lifecycle.OnAdded;
-import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
@@ -55,7 +56,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
private static final Logger logger = LoggerFactory.getLogger(StandardControllerServiceProvider.class);
- private final Map<String, ControllerServiceNode> controllerServices;
+ private final ConcurrentMap<String, ControllerServiceNode> controllerServices;
private static final Set<Method> validDisabledMethods;
static {
@@ -245,4 +246,9 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
controllerServices.remove(serviceNode.getIdentifier());
}
+
+ @Override
+ public Set<ControllerServiceNode> getAllControllerServices() {
+ return new HashSet<>(controllerServices.values());
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java
index c11aa72..039b2c2 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/persistence/StandardXMLFlowConfigurationDAO.java
@@ -219,7 +219,6 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD
//get properties for the specific reporting task - id, name, class,
//and schedulingPeriod must be set
- final String taskId = DomUtils.getChild(taskElement, "id").getTextContent().trim();
final String taskName = DomUtils.getChild(taskElement, "name").getTextContent().trim();
final List<Element> schedulingStrategyNodeList = DomUtils.getChildElementsByTagName(taskElement, "schedulingStrategy");
@@ -230,7 +229,7 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD
try {
schedulingStrategyValue = SchedulingStrategy.valueOf(specifiedValue).name();
} catch (final Exception e) {
- throw new RuntimeException("Cannot start Reporting Task with id " + taskId + " because its Scheduling Strategy does not have a valid value", e);
+ throw new RuntimeException("Cannot start Reporting Task with name " + taskName + " because its Scheduling Strategy does not have a valid value", e);
}
}
@@ -248,9 +247,9 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD
//set the class to be used for the configured reporting task
final ReportingTaskNode reportingTaskNode;
try {
- reportingTaskNode = controller.createReportingTask(taskClass, taskId);
+ reportingTaskNode = controller.createReportingTask(taskClass);
} catch (final ReportingTaskInstantiationException e) {
- LOG.error("Unable to load reporting task {} due to {}", new Object[]{taskId, e});
+ LOG.error("Unable to load reporting task {} due to {}", new Object[]{taskName, e});
if (LOG.isDebugEnabled()) {
LOG.error("", e);
}
@@ -263,7 +262,8 @@ public final class StandardXMLFlowConfigurationDAO implements FlowConfigurationD
final ReportingTask reportingTask = reportingTaskNode.getReportingTask();
- final ReportingInitializationContext config = new StandardReportingInitializationContext(taskId, taskName, schedulingStrategy, taskSchedulingPeriod, controller);
+ final ReportingInitializationContext config = new StandardReportingInitializationContext(
+ reportingTask.getIdentifier(), taskName, schedulingStrategy, taskSchedulingPeriod, controller);
reportingTask.initialize(config);
final Map<PropertyDescriptor, String> resolvedProps;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/86d15f9e/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index 603c1d2..5ed25e6 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@ -438,6 +438,10 @@ public class ControllerFacade implements ControllerServiceProvider {
return flowController.getControllerServiceName(serviceIdentifier);
}
+ @Override
+ public Set<ControllerServiceNode> getAllControllerServices() {
+ return flowController.getAllControllerServices();
+ }
@Override
public void enableControllerService(final ControllerServiceNode serviceNode) {