You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "Mihhai (via GitHub)" <gi...@apache.org> on 2025/09/26 13:48:59 UTC
[PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Mihhai opened a new pull request, #10353:
URL: https://github.com/apache/nifi/pull/10353
<!-- 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. -->
# Summary
[NIFI-15017](https://issues.apache.org/jira/browse/NIFI-15017)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
### Pull Request Formatting
- [x] Pull Request based on current revision of the `main` branch
- [x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [x] Build completed using `./mvnw clean install -P contrib-check`
- [x] JDK 21
- [ ] JDK 25
### Licensing
- [x] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [x] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [x] Documentation formatting appears as expected in rendered files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365952181
> > I believe it's currently not compatible with the API implemented by MiNiFi C++, but we're open to an API change if we can agree that the new API is an improvement, and if we can provide a transition period. I find it difficult to understand what protocol API this code maps to. In MiNiFi C++, I believe we have this:
> > ```
> > * Operation: START/STOP
> > valid "name": C2 | FlowController | <component UUID> | <component name>
> > content / arguments: (empty, not used)
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > From what I understand after talking to @pkedvessy, this pull request adds a new "name" called "processor, where we would put the processor UUID in the (currently unused) arguments, as a JSON Object with the key "processorId".
>
> Just to have a clear example currently in MiNiFi C++ the format of the start/stop request looks like the following where the operand is a processor id:
>
> ```
> {
> "operation": "heartbeat",
> "requested_operations": [
> {
> "operationid": 41,
> "operation": "stop",
> "operand": "2438e3c8-015a-1000-79ca-83af40ec1991"
> }
> ]
> }
> ```
>
> What is the equivalent json in this case?
Currently the json for a stop flow opearation which is already integrated in the java solution would be :
```
{
"requested_operations": [
{
"identifier": <some-id>,
"operation": "STOP",
"operand": "FLOW"
}
]
}
```
I believe this stops everything inside the root process group, and if anything other than FLOW is sent as operand the command fails
The json for the stop processor command i proposed would look like this:
```
{
"requested_operations": [
{
"identifier": <some-id>,
"operation": "STOP",
"operand": "PROCESSOR",
"args": {
"processorId": "processor_id"
}
}
]
}
```
This is the implementation i naturally gravitated towards given the selection machanism of handlers that the java implementation has. It selects the proper handler based on a combination of operation/operand.
I believe there are already some differences between the interfaces of the 2 implementations.So in order to be compatible with the cpp variant and have start/stop per processor we would need some sort of detection logic for the value received in the operand field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2401593899
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
Review Comment:
Nitpick:
Controlling the state transition with a boolean flag works but not well readable without checking the body of the method. I can see the reason behind: this way you did not duplicate the boilerplate code around the flowController calls. I think it would be a bit more readable by passing the state transition as a method reference:
Eg.: changeState(processorId, this::star) and changeState(processorId, this::stop)
private void start(String processorId, String parentGroupId) ...
private void stop(String processorId, String parentGroupId) ...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3380653825
Yes, It sounds like a bug. Please file a JIRA and leave it as it is now in this PR. The dedicated ticket and PR will give us better tracking on the bug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "lordgamez (via GitHub)" <gi...@apache.org>.
lordgamez commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2401636699
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
+ try {
+ ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
+ if (node == null) {
+ LOGGER.warn("Processor with id {} not found", processorId);
+ return NOT_APPLIED;
+ }
+ String parentGroupId = node.getProcessGroupIdentifier();
+ if (start) {
+ flowController.startProcessor(parentGroupId, processorId, true);
+ LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+ } else {
+ flowController.stopProcessor(parentGroupId, processorId);
+ LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+ }
+ return FULLY_APPLIED;
+ } catch (Exception e) {
+ LOGGER.error("Failed to change state for processor {}", processorId, e);
+ return PARTIALLY_APPLIED;
Review Comment:
@pkedvessy sorry I could not check the change earlier, @Mihhai is right we only respond with `FULLY_APPLIED` every time for the start/stop messages, thanks for adjusting the code to minificpp!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "szaszm (via GitHub)" <gi...@apache.org>.
szaszm commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3366023246
I agree with Gábor: we can go forward with this API and change minifi c++ and ask adamdebreceni/c2-server to use this new API. It seems to work better in the Java world to have the operand name be a static string.
@Mihhai just out of curiosity, what C2 server do you use? Do you have a custom one, or did you patch something to add this capability?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365364066
In general I'm happy with your PR but before moving forward to merge it I'd have someone from the CPP Agent team to make sure the C2 API is 100% compatible with their implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2398041499
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
+ try {
+ ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
+ if (node == null) {
+ LOGGER.warn("Processor with id {} not found", processorId);
+ return NOT_APPLIED;
+ }
+ String parentGroupId = node.getProcessGroupIdentifier();
+ if (start) {
+ flowController.startProcessor(parentGroupId, processorId, true);
+ LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+ } else {
+ flowController.stopProcessor(parentGroupId, processorId);
+ LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+ }
+ return FULLY_APPLIED;
+ } catch (Exception e) {
+ LOGGER.error("Failed to change state for processor {}", processorId, e);
+ return PARTIALLY_APPLIED;
Review Comment:
According to the cpp implementation ( [response construction for start/stop](https://github.com/apache/nifi-minifi-cpp/blob/main/libminifi/src/c2/C2Agent.cpp#L360) and [c2payload implementation](https://github.com/apache/nifi-minifi-cpp/blob/main/libminifi/src/c2/C2Payload.cpp#L28) ) it seems the response to a successful start/stop is a `FULLY_APPLIED` status. As a result i removed the `PARTIALLY_APPLIED ` response as you suggested but still left the `NOT_APPLIED` in case of error. From my understanding of the cpp code there doesnt seem to be any handling of the case where the start/stop operation fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "lordgamez (via GitHub)" <gi...@apache.org>.
lordgamez commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365606622
> I believe it's currently not compatible with the API implemented by MiNiFi C++, but we're open to an API change if we can agree that the new API is an improvement, and if we can provide a transition period. I find it difficult to understand what protocol API this code maps to. In MiNiFi C++, I believe we have this:
>
> * Operation: START/STOP
> valid "name": C2 | FlowController | <component UUID> | <component name>
> content / arguments: (empty, not used)
>
>
> From what I understand after talking to @pkedvessy, this pull request adds a new "name" called "processor, where we would put the processor UUID in the (currently unused) arguments, as a JSON Object with the key "processorId".
Just to have a clear example currently in MiNiFi C++ the format of the start/stop request looks like the following where the operand is a processor id:
```
{
"operation": "heartbeat",
"requested_operations": [
{
"operationid": 41,
"operation": "stop",
"operand": "2438e3c8-015a-1000-79ca-83af40ec1991"
}
]
}
```
What is the equivalent json in this case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3366120885
> I agree with Gábor: we can go forward with this API and change minifi c++ and ask adamdebreceni/c2-server to use this new API. It seems to work better in the Java world to have the operand name be a static string.
>
> @Mihhai just out of curiosity, what C2 server do you use? Do you have a custom one, or did you patch something to add this capability? If it's open source, we may want to add integration tests with it to ensure no breakage in the future.
I just implemented a small, custom server to exercise and validate the flows during development. Nothing special, i just exposed 2 endpoints (heartbeat and acknowledgment) and some simple mechanism to queue up commands
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "szaszm (via GitHub)" <gi...@apache.org>.
szaszm commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365594055
I believe it's currently not compatible with the API implemented by MiNiFi C++, but we're open to an API change if we can agree that the new API is an improvement, and if we can provide a transition period. I find it difficult to understand what protocol API this code maps to. In MiNiFi C++, I believe we have this:
- Operation: START/STOP
valid "name": C2 | FlowController | \<component UUID\> | \<component name\>
content / arguments: (empty, not used)
From what I understand after talking to @pkedvessy, this pull request adds a new "name" called "processor, where we would put the processor UUID in the (currently unused) arguments, as a JSON Object with the key "processorId".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2409594590
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
Review Comment:
Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "lordgamez (via GitHub)" <gi...@apache.org>.
lordgamez commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365959944
> Currently the json for a stop flow opearation which is already integrated in the java solution would be :
>
> ```
> {
> "requested_operations": [
> {
> "identifier": <some-id>,
> "operation": "STOP",
> "operand": "FLOW"
> }
> ]
> }
> ```
>
> I believe this stops everything inside the root process group, and if anything other than FLOW is sent as operand the command fails
>
> The json for the stop processor command i proposed would look like this:
>
> ```
> {
> "requested_operations": [
> {
> "identifier": <some-id>,
> "operation": "STOP",
> "operand": "PROCESSOR",
> "args": {
> "processorId": "processor_id"
> }
> }
> ]
> }
> ```
>
> This is the implementation i naturally gravitated towards given the selection machanism of handlers that the java implementation has. It selects the proper handler based on a combination of operation/operand.
>
> I believe there are already some differences between the interfaces of the 2 implementations.So in order to be compatible with the cpp variant and have start/stop per processor we would need some sort of detection logic for the value received in the operand field.
I think this is okay to move forward with this implementation and we'll adjust minifi-cpp and the open source minifi-c2 server implementation to this in the future. It will be easier to accept both the old and the new notation in minifi-cpp for a while until the old notation is taken out in a future release.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365919045
Currently the json for a stop flow opearation which is already integrated in the solution would be :
`{
"requested_operations": [
{
"identifier": <some-id>,
"operation": "STOP",
"operand": "FLOW"
}
]
}`
I believe this stops everything inside the root process group, and if anything other than FLOW is sent as operand the command fails
The json for the stop processor command i proposed would look like this:
`{
"requested_operations": [
{
"identifier": <some-id>,
"operation": "STOP",
"operand": "PROCESSOR",
"args": {
"processorId": "processor_id"
}
}
]
}`
This is the implementation i naturally gravitated towards given the selection machanism of handlers that the java implementation has. It selects the proper handler based on a combination of operation/operand.
I believe there are already some differences between the interfaces of the 2 implementations.So in order to be compatible with the cpp variant and have start/stop per processor we would need some sort of detection logic for the value received in the operand field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2398008061
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java:
##########
@@ -427,9 +435,29 @@ private ProcessorStatus convertProcessorStatus(org.apache.nifi.controller.status
result.setProcessingNanos(processorStatus.getProcessingNanos());
result.setActiveThreadCount(processorStatus.getActiveThreadCount());
result.setTerminatedThreadCount(processorStatus.getTerminatedThreadCount());
+ RunStatus mappedRunStatus = mapProcessorRunStatus(processorStatus.getRunStatus());
+ if (mappedRunStatus != null) {
+ result.setRunStatus(mappedRunStatus);
+ }
return result;
}
+ private RunStatus mapProcessorRunStatus(org.apache.nifi.controller.status.RunStatus controllerRunStatus) {
Review Comment:
i refactor to the one line mapping you suggested and i removed the helper method.
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java:
##########
@@ -427,9 +435,29 @@ private ProcessorStatus convertProcessorStatus(org.apache.nifi.controller.status
result.setProcessingNanos(processorStatus.getProcessingNanos());
result.setActiveThreadCount(processorStatus.getActiveThreadCount());
result.setTerminatedThreadCount(processorStatus.getTerminatedThreadCount());
+ RunStatus mappedRunStatus = mapProcessorRunStatus(processorStatus.getRunStatus());
+ if (mappedRunStatus != null) {
+ result.setRunStatus(mappedRunStatus);
+ }
return result;
}
+ private RunStatus mapProcessorRunStatus(org.apache.nifi.controller.status.RunStatus controllerRunStatus) {
Review Comment:
i refactored to the one line mapping you suggested and i removed the helper method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "adamdebreceni (via GitHub)" <gi...@apache.org>.
adamdebreceni commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3375759847
good change, don't worry about the c2-server we'll update accordingly, does this also add the processor status to the heartbeat, or was that already included?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2401597206
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
+ try {
+ ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
+ if (node == null) {
+ LOGGER.warn("Processor with id {} not found", processorId);
+ return NOT_APPLIED;
+ }
+ String parentGroupId = node.getProcessGroupIdentifier();
+ if (start) {
+ flowController.startProcessor(parentGroupId, processorId, true);
+ LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+ } else {
+ flowController.stopProcessor(parentGroupId, processorId);
+ LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+ }
+ return FULLY_APPLIED;
+ } catch (Exception e) {
+ LOGGER.error("Failed to change state for processor {}", processorId, e);
+ return PARTIALLY_APPLIED;
Review Comment:
Thx
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java:
##########
@@ -427,9 +435,29 @@ private ProcessorStatus convertProcessorStatus(org.apache.nifi.controller.status
result.setProcessingNanos(processorStatus.getProcessingNanos());
result.setActiveThreadCount(processorStatus.getActiveThreadCount());
result.setTerminatedThreadCount(processorStatus.getTerminatedThreadCount());
+ RunStatus mappedRunStatus = mapProcessorRunStatus(processorStatus.getRunStatus());
+ if (mappedRunStatus != null) {
+ result.setRunStatus(mappedRunStatus);
+ }
return result;
}
+ private RunStatus mapProcessorRunStatus(org.apache.nifi.controller.status.RunStatus controllerRunStatus) {
Review Comment:
Thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3365940261
> > I believe it's currently not compatible with the API implemented by MiNiFi C++, but we're open to an API change if we can agree that the new API is an improvement, and if we can provide a transition period. I find it difficult to understand what protocol API this code maps to. In MiNiFi C++, I believe we have this:
> > ```
> > * Operation: START/STOP
> > valid "name": C2 | FlowController | <component UUID> | <component name>
> > content / arguments: (empty, not used)
> > ```
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > From what I understand after talking to @pkedvessy, this pull request adds a new "name" called "processor, where we would put the processor UUID in the (currently unused) arguments, as a JSON Object with the key "processorId".
>
> Just to have a clear example currently in MiNiFi C++ the format of the start/stop request looks like the following where the operand is a processor id:
>
> ```
> {
> "operation": "heartbeat",
> "requested_operations": [
> {
> "operationid": 41,
> "operation": "stop",
> "operand": "2438e3c8-015a-1000-79ca-83af40ec1991"
> }
> ]
> }
> ```
>
> What is the equivalent json in this case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai closed pull request #10353: NIFI-15017 Add start/stop C2 command for processors in MiNiFi
URL: https://github.com/apache/nifi/pull/10353
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2409590501
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
Review Comment:
refactored as suggested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2389332613
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
+ try {
+ ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
+ if (node == null) {
+ LOGGER.warn("Processor with id {} not found", processorId);
+ return NOT_APPLIED;
+ }
+ String parentGroupId = node.getProcessGroupIdentifier();
+ if (start) {
+ flowController.startProcessor(parentGroupId, processorId, true);
+ LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+ } else {
+ flowController.stopProcessor(parentGroupId, processorId);
+ LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+ }
+ return FULLY_APPLIED;
+ } catch (Exception e) {
+ LOGGER.error("Failed to change state for processor {}", processorId, e);
+ return PARTIALLY_APPLIED;
Review Comment:
In my opinion the PARTIALLY_APPLIED result is not suitable for a single processor start/stop operation. It would make sense for a processor group what could be partially started or stopped (eg some of the processors in the group are started or stopped by the operation meanwhile others remain in the opposite state for some reason) but the operation result for a single processor should not be partial.
@lordgamez could you please check what could be the operation state from the CPP agent in case of a processor start/stop operation? We should keep this in line with the existing implementation in the CPP agent.
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java:
##########
@@ -427,9 +435,29 @@ private ProcessorStatus convertProcessorStatus(org.apache.nifi.controller.status
result.setProcessingNanos(processorStatus.getProcessingNanos());
result.setActiveThreadCount(processorStatus.getActiveThreadCount());
result.setTerminatedThreadCount(processorStatus.getTerminatedThreadCount());
+ RunStatus mappedRunStatus = mapProcessorRunStatus(processorStatus.getRunStatus());
+ if (mappedRunStatus != null) {
+ result.setRunStatus(mappedRunStatus);
+ }
return result;
}
+ private RunStatus mapProcessorRunStatus(org.apache.nifi.controller.status.RunStatus controllerRunStatus) {
Review Comment:
In the switch statement, we can simply return STOPPED as default and we can keep RUNNING mapping as it is now.
In other hand, controllerRunStatus parameter naming is misleading. We should use something like a runStatus or processorRunStatus as parameter name or we can simply set the runStatus without this method like
`result.setRunStatus(processorStatus.getRunStatus() == Running ? RUNNING : STOPPED);`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3380486426
> good change, don't worry about the c2-server we'll update accordingly, does this also add the processor status to the heartbeat, or was that already included?
Yes it does, and while we're at it i'd also like to mention that [looking at the current implementation](https://github.com/apache/nifi/blob/main/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java#L404) i noticed the heartbeat’s FlowInfo seems to include only processors and connection queues that are direct children of the root process group. In my tests, if the root group only contains sub process groups (and those hold the actual processors/queues), those inner components do not appear in the heartbeat payload. I assume this isnt the intended behaviour, should i treat this as a bug? If so, should i sneak in a fix for this in this pr also?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "Mihhai (via GitHub)" <gi...@apache.org>.
Mihhai commented on code in PR #10353:
URL: https://github.com/apache/nifi/pull/10353#discussion_r2398041499
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultProcessorStateStrategy.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.minifi.c2.command;
+
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.client.service.operation.ProcessorStateStrategy;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.ProcessorNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+public class DefaultProcessorStateStrategy implements ProcessorStateStrategy {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessorStateStrategy.class);
+
+ private final FlowController flowController;
+
+ public DefaultProcessorStateStrategy(FlowController flowController) {
+ this.flowController = flowController;
+ }
+
+ @Override
+ public OperationState startProcessor(String processorId) {
+ return changeState(processorId, true);
+ }
+
+ @Override
+ public OperationState stopProcessor(String processorId) {
+ return changeState(processorId, false);
+ }
+
+ private OperationState changeState(String processorId, boolean start) {
+ try {
+ ProcessorNode node = flowController.getFlowManager().getProcessorNode(processorId);
+ if (node == null) {
+ LOGGER.warn("Processor with id {} not found", processorId);
+ return NOT_APPLIED;
+ }
+ String parentGroupId = node.getProcessGroupIdentifier();
+ if (start) {
+ flowController.startProcessor(parentGroupId, processorId, true);
+ LOGGER.info("Started processor {} (group={})", processorId, parentGroupId);
+ } else {
+ flowController.stopProcessor(parentGroupId, processorId);
+ LOGGER.info("Stopped processor {} (group={})", processorId, parentGroupId);
+ }
+ return FULLY_APPLIED;
+ } catch (Exception e) {
+ LOGGER.error("Failed to change state for processor {}", processorId, e);
+ return PARTIALLY_APPLIED;
Review Comment:
According to the cpp implementation [response construction for start/stop](https://github.com/apache/nifi-minifi-cpp/blob/main/libminifi/src/c2/C2Agent.cpp#L360) and [c2payload implementation](https://github.com/apache/nifi-minifi-cpp/blob/main/libminifi/src/c2/C2Payload.cpp#L28) it seems the response to a successful start/stop is a `FULLY_APPLIED` status. As a result i removed the `PARTIALLY_APPLIED ` response as you suggested but still left the `NOT_APPLIED` in case of error. From my understanding of the cpp code there doesnt seem to be any handling of the case where the start/stop operation fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "szaszm (via GitHub)" <gi...@apache.org>.
szaszm closed pull request #10353: NIFI-15017 Add start/stop C2 command for processors in MiNiFi
URL: https://github.com/apache/nifi/pull/10353
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15017 Add start/stop C2 command for processors in MiNiFi [nifi]
Posted by "pkedvessy (via GitHub)" <gi...@apache.org>.
pkedvessy commented on PR #10353:
URL: https://github.com/apache/nifi/pull/10353#issuecomment-3429759390
Thanks @szaszm for taking care about this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org