You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2017/05/16 20:04:37 UTC
nifi git commit: NIFI-3900: Ensure that when we serialize a flow to
send to Cluster Coordinator that we include the Scheduled State for
processors as they are intended to be,
not as they are currently because during startup they may not have been
started
Repository: nifi
Updated Branches:
refs/heads/master ac8e57259 -> 494a0e892
NIFI-3900: Ensure that when we serialize a flow to send to Cluster Coordinator that we include the Scheduled State for processors as they are intended to be, not as they are currently because during startup they may not have been started yet.
NIFI-3900: Rebased against master and updated new unit test to use new method signature for FlowSerializer.serialize
This closes #1804.
Signed-off-by: Bryan Bende <bb...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/494a0e89
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/494a0e89
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/494a0e89
Branch: refs/heads/master
Commit: 494a0e89288c5e51f9467b719e7ee582cd74c875
Parents: ac8e572
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon May 15 15:38:07 2017 -0400
Committer: Bryan Bende <bb...@apache.org>
Committed: Tue May 16 16:04:09 2017 -0400
----------------------------------------------------------------------
.../apache/nifi/controller/FlowController.java | 6 ++++-
.../controller/StandardFlowSynchronizer.java | 2 ++
.../serialization/FlowSerializer.java | 3 ++-
.../serialization/ScheduledStateLookup.java | 28 ++++++++++++++++++++
.../serialization/StandardFlowSerializer.java | 14 +++++-----
.../controller/StandardFlowServiceTest.java | 9 ++++---
.../StandardFlowSerializerTest.java | 2 +-
7 files changed, 50 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 5b2ea41..3ca28e5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -138,6 +138,7 @@ import org.apache.nifi.controller.serialization.FlowSerializationException;
import org.apache.nifi.controller.serialization.FlowSerializer;
import org.apache.nifi.controller.serialization.FlowSynchronizationException;
import org.apache.nifi.controller.serialization.FlowSynchronizer;
+import org.apache.nifi.controller.serialization.ScheduledStateLookup;
import org.apache.nifi.controller.service.ControllerServiceInvocationHandler;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
@@ -1509,7 +1510,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
public void serialize(final FlowSerializer serializer, final OutputStream os) throws FlowSerializationException {
readLock.lock();
try {
- serializer.serialize(this, os);
+ final ScheduledStateLookup scheduledStateLookup = procNode -> startConnectablesAfterInitialization.contains(procNode) ? ScheduledState.RUNNING : procNode.getScheduledState();
+ serializer.serialize(this, os, scheduledStateLookup);
} finally {
readLock.unlock();
}
@@ -2932,6 +2934,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
throw new IllegalStateException("Cannot find ProcessorNode with ID " + processorId + " within ProcessGroup with ID " + parentGroupId);
}
group.stopProcessor(node);
+ // If we are ready to start the processor upon initialization of the controller, don't.
+ startConnectablesAfterInitialization.remove(node);
}
public void stopAllProcessors() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
index 1e229a2..a9372a4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSynchronizer.java
@@ -1006,6 +1006,8 @@ public class StandardFlowSynchronizer implements FlowSynchronizer {
controller.startProcessor(processGroup.getIdentifier(), procNode.getIdentifier());
} else if (ScheduledState.DISABLED.equals(scheduledState)) {
processGroup.disableProcessor(procNode);
+ } else if (ScheduledState.STOPPED.equals(scheduledState)) {
+ controller.stopProcessor(processGroup.getIdentifier(), procNode.getIdentifier());
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowSerializer.java
index dae7566..b01f6e3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowSerializer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/FlowSerializer.java
@@ -34,9 +34,10 @@ public interface FlowSerializer {
*
* @param controller a controller
* @param os an output stream to write the configuration to
+ * @param stateLookup a lookup that can be used to determine the ScheduledState of a Processor
*
* @throws FlowSerializationException if serialization failed
*/
- void serialize(FlowController controller, OutputStream os) throws FlowSerializationException;
+ void serialize(FlowController controller, OutputStream os, ScheduledStateLookup stateLookup) throws FlowSerializationException;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/ScheduledStateLookup.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/ScheduledStateLookup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/ScheduledStateLookup.java
new file mode 100644
index 0000000..07f6017
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/ScheduledStateLookup.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.controller.serialization;
+
+import org.apache.nifi.controller.ProcessorNode;
+import org.apache.nifi.controller.ScheduledState;
+
+public interface ScheduledStateLookup {
+
+ ScheduledState getScheduledState(ProcessorNode procNode);
+
+ public static final ScheduledStateLookup IDENTITY_LOOKUP = ProcessorNode::getScheduledState;
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
index cd02769..fea1ecb 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/StandardFlowSerializer.java
@@ -79,7 +79,7 @@ public class StandardFlowSerializer implements FlowSerializer {
}
@Override
- public void serialize(final FlowController controller, final OutputStream os) throws FlowSerializationException {
+ public void serialize(final FlowController controller, final OutputStream os, final ScheduledStateLookup scheduledStateLookup) throws FlowSerializationException {
try {
// create a new, empty document
final DocumentBuilderFactory docFactory = DocumentBuilderFactory.newInstance();
@@ -94,7 +94,7 @@ public class StandardFlowSerializer implements FlowSerializer {
doc.appendChild(rootNode);
addTextElement(rootNode, "maxTimerDrivenThreadCount", controller.getMaxTimerDrivenThreadCount());
addTextElement(rootNode, "maxEventDrivenThreadCount", controller.getMaxEventDrivenThreadCount());
- addProcessGroup(rootNode, controller.getGroup(controller.getRootGroupId()), "rootGroup");
+ addProcessGroup(rootNode, controller.getGroup(controller.getRootGroupId()), "rootGroup", scheduledStateLookup);
// Add root-level controller services
final Element controllerServicesNode = doc.createElement("controllerServices");
@@ -144,7 +144,7 @@ public class StandardFlowSerializer implements FlowSerializer {
parentElement.appendChild(element);
}
- private void addProcessGroup(final Element parentElement, final ProcessGroup group, final String elementName) {
+ private void addProcessGroup(final Element parentElement, final ProcessGroup group, final String elementName, final ScheduledStateLookup scheduledStateLookup) {
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement(elementName);
parentElement.appendChild(element);
@@ -154,7 +154,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "comment", group.getComments());
for (final ProcessorNode processor : group.getProcessors()) {
- addProcessor(element, processor);
+ addProcessor(element, processor, scheduledStateLookup);
}
if (group.isRootGroup()) {
@@ -184,7 +184,7 @@ public class StandardFlowSerializer implements FlowSerializer {
}
for (final ProcessGroup childGroup : group.getProcessGroups()) {
- addProcessGroup(element, childGroup, "processGroup");
+ addProcessGroup(element, childGroup, "processGroup", scheduledStateLookup);
}
for (final RemoteProcessGroup remoteRef : group.getRemoteProcessGroups()) {
@@ -363,7 +363,7 @@ public class StandardFlowSerializer implements FlowSerializer {
parentElement.appendChild(element);
}
- private void addProcessor(final Element parentElement, final ProcessorNode processor) {
+ private void addProcessor(final Element parentElement, final ProcessorNode processor, final ScheduledStateLookup scheduledStateLookup) {
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement("processor");
parentElement.appendChild(element);
@@ -384,7 +384,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "yieldPeriod", processor.getYieldPeriod());
addTextElement(element, "bulletinLevel", processor.getBulletinLevel().toString());
addTextElement(element, "lossTolerant", String.valueOf(processor.isLossTolerant()));
- addTextElement(element, "scheduledState", processor.getScheduledState().name());
+ addTextElement(element, "scheduledState", scheduledStateLookup.getScheduledState(processor).name());
addTextElement(element, "schedulingStrategy", processor.getSchedulingStrategy().name());
addTextElement(element, "executionNode", processor.getExecutionNode().name());
addTextElement(element, "runDurationNanos", processor.getRunDuration(TimeUnit.NANOSECONDS));
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java
index 0bc216f..1957cf4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/StandardFlowServiceTest.java
@@ -23,6 +23,7 @@ import org.apache.nifi.cluster.protocol.StandardDataFlow;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.serialization.FlowSerializationException;
import org.apache.nifi.controller.serialization.FlowSerializer;
+import org.apache.nifi.controller.serialization.ScheduledStateLookup;
import org.apache.nifi.controller.serialization.StandardFlowSerializer;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.events.VolatileBulletinRepository;
@@ -96,7 +97,7 @@ public class StandardFlowServiceTest {
FlowSerializer serializer = new StandardFlowSerializer(mockEncryptor);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- serializer.serialize(flowController, baos);
+ serializer.serialize(flowController, baos, ScheduledStateLookup.IDENTITY_LOOKUP);
String expectedFlow = new String(flowBytes).trim();
String actualFlow = new String(baos.toByteArray()).trim();
@@ -120,7 +121,7 @@ public class StandardFlowServiceTest {
FlowSerializer serializer = new StandardFlowSerializer(mockEncryptor);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- serializer.serialize(flowController, baos);
+ serializer.serialize(flowController, baos, ScheduledStateLookup.IDENTITY_LOOKUP);
String expectedFlow = new String(flowBytes).trim();
String actualFlow = new String(baos.toByteArray()).trim();
@@ -140,7 +141,7 @@ public class StandardFlowServiceTest {
FlowSerializer serializer = new StandardFlowSerializer(mockEncryptor);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- serializer.serialize(flowController, baos);
+ serializer.serialize(flowController, baos, ScheduledStateLookup.IDENTITY_LOOKUP);
String expectedFlow = new String(originalBytes).trim();
String actualFlow = new String(baos.toByteArray()).trim();
@@ -162,7 +163,7 @@ public class StandardFlowServiceTest {
FlowSerializer serializer = new StandardFlowSerializer(mockEncryptor);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- serializer.serialize(flowController, baos);
+ serializer.serialize(flowController, baos, ScheduledStateLookup.IDENTITY_LOOKUP);
String expectedFlow = new String(originalBytes).trim();
String actualFlow = new String(baos.toByteArray()).trim();
http://git-wip-us.apache.org/repos/asf/nifi/blob/494a0e89/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/StandardFlowSerializerTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/StandardFlowSerializerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/StandardFlowSerializerTest.java
index e85edb0..f61dcc2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/StandardFlowSerializerTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/StandardFlowSerializerTest.java
@@ -92,7 +92,7 @@ public class StandardFlowSerializerTest {
// serialize the controller
final ByteArrayOutputStream os = new ByteArrayOutputStream();
- serializer.serialize(controller, os);
+ serializer.serialize(controller, os, ScheduledStateLookup.IDENTITY_LOOKUP);
// verify the results contain the serialized string
final String serializedFlow = os.toString(StandardCharsets.UTF_8.name());