briansolo1985 commented on code in PR #9694: URL: https://github.com/apache/nifi/pull/9694#discussion_r1944764999
########## c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/StartFlowOperationHandlerTest.java: ########## @@ -0,0 +1,95 @@ +/* + * 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.c2.client.service.operation; + +import java.util.stream.Stream; +import org.apache.nifi.c2.protocol.api.C2Operation; +import org.apache.nifi.c2.protocol.api.C2OperationAck; +import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.apache.commons.lang3.StringUtils.EMPTY; +import static org.apache.nifi.c2.client.service.operation.StartFlowOperationHandler.FULLY_APPLIED_DETAILS; +import static org.apache.nifi.c2.client.service.operation.StartFlowOperationHandler.NOT_APPLIED_DETAILS; +import static org.apache.nifi.c2.client.service.operation.StartFlowOperationHandler.PARTIALLY_APPLIED_DETAILS; +import static org.apache.nifi.c2.client.service.operation.StartFlowOperationHandler.UNEXPECTED_DETAILS; +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.NO_OPERATION; +import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED; +import static org.apache.nifi.c2.protocol.api.OperandType.FLOW; +import static org.apache.nifi.c2.protocol.api.OperationType.START; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +public class StartFlowOperationHandlerTest { Review Comment: Thanks for the thorough testing ########## c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2HeartbeatFactory.java: ########## @@ -102,11 +98,12 @@ public C2Heartbeat create(RuntimeInfoWrapper runtimeInfoWrapper) { return heartbeat; } - private FlowInfo getFlowInfo(Map<String, FlowQueueStatus> queueStatus, List<ProcessorBulletin> processorBulletins, List<ProcessorStatus> processorStatus) { + private FlowInfo getFlowInfo(RuntimeInfoWrapper runtimeInfoWrapper) { Review Comment: Thanks, looks better this way ########## c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/StopFlowOperationHandler.java: ########## @@ -0,0 +1,78 @@ +/* + * 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.c2.client.service.operation; + +import org.apache.nifi.c2.protocol.api.C2Operation; +import org.apache.nifi.c2.protocol.api.C2OperationAck; +import org.apache.nifi.c2.protocol.api.C2OperationState; +import org.apache.nifi.c2.protocol.api.OperandType; +import org.apache.nifi.c2.protocol.api.OperationType; + +import java.util.Collections; +import java.util.Map; + +import static java.util.Optional.ofNullable; +import static org.apache.commons.lang3.StringUtils.EMPTY; + +public class StopFlowOperationHandler implements C2OperationHandler { + + public static final String NOT_APPLIED_DETAILS = "Failed to stop flow, please check the log for errors"; + public static final String FULLY_APPLIED_DETAILS = "Flow stopped"; + public static final String PARTIALLY_APPLIED_DETAILS = "Some components failed to stop, please check the log for errors"; + public static final String UNEXPECTED_DETAILS = "Unexpected status, please check the log for errors"; + Review Comment: We could remove one empty line ########## minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java: ########## @@ -150,13 +158,11 @@ public C2NifiClientService(NiFiProperties niFiProperties, FlowController flowCon this.heartbeatManagerExecutorService = newScheduledThreadPool(1); this.operationManagerExecutorService = newSingleThreadExecutor(); - this.extensionManifestParser = new JAXBExtensionManifestParser(); - C2ClientConfig clientConfig = generateClientConfig(niFiProperties); this.runtimeManifestService = new StandardRuntimeManifestService( ExtensionManagerHolder.getExtensionManager(), - extensionManifestParser, + new JAXBExtensionManifestParser(), Review Comment: Good catch, this is not needed to be an instance variable. Please correct the formatting ########## minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultFlowStateStrategy.java: ########## @@ -0,0 +1,91 @@ +/* + * 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.client.service.operation.FlowStateStrategy; +import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState; +import org.apache.nifi.controller.FlowController; +import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.groups.RemoteProcessGroup; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +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 DefaultFlowStateStrategy implements FlowStateStrategy { + + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultFlowStateStrategy.class); + + private final FlowController flowController; + + public DefaultFlowStateStrategy(FlowController flowController) { + this.flowController = flowController; + } + + @Override + public OperationState start() { + try { + ProcessGroup rootProcessGroup = flowController.getFlowManager().getRootGroup(); + if (rootProcessGroup != null) { + rootProcessGroup.startProcessing(); Review Comment: If I remember correctly processGroup.startProcessing() starts the underlying process groups as well, so that seems to be okay. However it seems to be that the remote process groups need to be started/stopped manually, so we probably need to traverse the process groups recursively, and start the remote process groups on each levels -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
