exceptionfactory commented on code in PR #6993: URL: https://github.com/apache/nifi/pull/6993#discussion_r1128484719
########## nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/logging/ParameterContextTaskLogObserver.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.logging; + +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.events.BulletinFactory; +import org.apache.nifi.reporting.Bulletin; +import org.apache.nifi.reporting.BulletinRepository; +import org.apache.nifi.reporting.ComponentType; +import org.apache.nifi.reporting.Severity; + +public class ParameterContextTaskLogObserver implements LogObserver { + private final BulletinRepository bulletinRepository; + private final ParameterProviderNode taskNode; Review Comment: This should be named `providerNode` instead of `taskNode` to align with the element being observed. ```suggestion private final ParameterProviderNode providerNode; ``` ########## nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/logging/ParameterContextTaskLogObserver.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.logging; + +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.events.BulletinFactory; +import org.apache.nifi.reporting.Bulletin; +import org.apache.nifi.reporting.BulletinRepository; +import org.apache.nifi.reporting.ComponentType; +import org.apache.nifi.reporting.Severity; + +public class ParameterContextTaskLogObserver implements LogObserver { + private final BulletinRepository bulletinRepository; + private final ParameterProviderNode taskNode; + + public ParameterContextTaskLogObserver(final BulletinRepository bulletinRepository, final ParameterProviderNode taskNode) { + this.bulletinRepository = bulletinRepository; + this.taskNode = taskNode; + } + + @Override + public void onLogMessage(final LogMessage message) { + // Map LogLevel.WARN to Severity.WARNING so that we are consistent with the Severity enumeration. Else, just use whatever + // the LogLevel is (INFO and ERROR map directly and all others we will just accept as they are). + final String bulletinLevel = message.getLogLevel() == LogLevel.WARN ? Severity.WARNING.name() : message.getLogLevel().toString(); + + final Bulletin bulletin = BulletinFactory.createBulletin(null, taskNode.getIdentifier(), ComponentType.REPORTING_TASK, Review Comment: The `ComponentType` should be `PARAMETER_PROVIDER` instead of `REPORTING_TASK`: ```suggestion final Bulletin bulletin = BulletinFactory.createBulletin(null, taskNode.getIdentifier(), ComponentType.PARAMETER_PROVIDER, ``` ########## nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/logging/ParameterContextTaskLogObserver.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.logging; + +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.events.BulletinFactory; +import org.apache.nifi.reporting.Bulletin; +import org.apache.nifi.reporting.BulletinRepository; +import org.apache.nifi.reporting.ComponentType; +import org.apache.nifi.reporting.Severity; + +public class ParameterContextTaskLogObserver implements LogObserver { + private final BulletinRepository bulletinRepository; + private final ParameterProviderNode taskNode; + + public ParameterContextTaskLogObserver(final BulletinRepository bulletinRepository, final ParameterProviderNode taskNode) { Review Comment: As above, `taskNode` should be renamed to `providerNode`: ```suggestion public ParameterContextTaskLogObserver(final BulletinRepository bulletinRepository, final ParameterProviderNode providerNode) { ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +320,54 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + requireNonNull(type); + requireNonNull(id); + requireNonNull(bundleCoordinate); Review Comment: The `requireNonNull` method takes a message, so recommend adding a message for each of these lines to clarify the problem. ```suggestion requireNonNull(type, "Provider Type required"); requireNonNull(id, "Provider ID required"); requireNonNull(bundleCoordinate, "Provider Bundle Coordinate required"); ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id + + ", bundle coordinate = " + bundleCoordinate); + } + + // make sure the first reference to LogRepository happens outside of a NarCloseable so that we use the framework's ClassLoader + final LogRepository logRepository = LogRepositoryFactory.getRepository(id); + + final ParameterProviderNode taskNode; + try { + taskNode = new ComponentBuilder() + .identifier(id) + .type(type) + .bundleCoordinate(bundleCoordinate) + .statelessEngine(statelessEngine) + .additionalClassPathUrls(additionalUrls) + .flowManager(this) + .buildParameterProvider(); + } catch (ParameterProviderInstantiationException e) { + throw new RuntimeException("Could not create Parameter Context Task of type " + type + " with ID " + id, e); Review Comment: Following the pattern of other methods, this should throw an `IllegalStateException` instead of the more generic `RuntimeException`. ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id Review Comment: The wording should be changed to indicate `Parameter provider` instead of `Reporting Task`: ```suggestion throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Parameter Provider. Provided arguments were type=" + type + ", id=" + id ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id + + ", bundle coordinate = " + bundleCoordinate); + } + + // make sure the first reference to LogRepository happens outside of a NarCloseable so that we use the framework's ClassLoader + final LogRepository logRepository = LogRepositoryFactory.getRepository(id); + + final ParameterProviderNode taskNode; Review Comment: The variable name should be changed to `providerNode` instead of `taskNode` to follow the naming convention of the class. ```suggestion final ParameterProviderNode providerNode; ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id + + ", bundle coordinate = " + bundleCoordinate); + } + + // make sure the first reference to LogRepository happens outside of a NarCloseable so that we use the framework's ClassLoader + final LogRepository logRepository = LogRepositoryFactory.getRepository(id); + + final ParameterProviderNode taskNode; + try { + taskNode = new ComponentBuilder() + .identifier(id) + .type(type) + .bundleCoordinate(bundleCoordinate) + .statelessEngine(statelessEngine) + .additionalClassPathUrls(additionalUrls) + .flowManager(this) + .buildParameterProvider(); + } catch (ParameterProviderInstantiationException e) { + throw new RuntimeException("Could not create Parameter Context Task of type " + type + " with ID " + id, e); + } - throw new UnsupportedOperationException("Parameter Providers are not supported in Stateless NiFi"); + LogRepositoryFactory.getRepository(taskNode.getIdentifier()).setLogger(taskNode.getLogger()); + + if (firstTimeAdded) { + final Class<?> taskClass = taskNode.getParameterProvider().getClass(); Review Comment: ```suggestion final Class<?> providerClass = taskNode.getParameterProvider().getClass(); ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id + + ", bundle coordinate = " + bundleCoordinate); + } + + // make sure the first reference to LogRepository happens outside of a NarCloseable so that we use the framework's ClassLoader + final LogRepository logRepository = LogRepositoryFactory.getRepository(id); + + final ParameterProviderNode taskNode; + try { + taskNode = new ComponentBuilder() + .identifier(id) + .type(type) + .bundleCoordinate(bundleCoordinate) + .statelessEngine(statelessEngine) + .additionalClassPathUrls(additionalUrls) + .flowManager(this) + .buildParameterProvider(); + } catch (ParameterProviderInstantiationException e) { + throw new RuntimeException("Could not create Parameter Context Task of type " + type + " with ID " + id, e); + } - throw new UnsupportedOperationException("Parameter Providers are not supported in Stateless NiFi"); + LogRepositoryFactory.getRepository(taskNode.getIdentifier()).setLogger(taskNode.getLogger()); + + if (firstTimeAdded) { + final Class<?> taskClass = taskNode.getParameterProvider().getClass(); + final String identifier = taskNode.getParameterProvider().getIdentifier(); + + try (final NarCloseable x = NarCloseable.withComponentNarLoader(statelessEngine.getExtensionManager(), taskClass, identifier)) { + ReflectionUtils.invokeMethodsWithAnnotation(OnAdded.class, taskNode.getParameterProvider()); + + if (isFlowInitialized()) { + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnConfigurationRestored.class, taskNode.getParameterProvider(), taskNode.getConfigurationContext()); + } + } catch (final Exception e) { + throw new ComponentLifeCycleException("Failed to invoke On-Added Lifecycle methods of " + taskNode.getParameterProvider(), e); + } + } + + if (registerLogObserver) { + onParameterProviderAdded(taskNode); + + // Register log observer to provide bulletins when reporting task logs anything at WARN level or above Review Comment: This comment appears to be copied from Reporting Tasks, so it should be updated or removed. ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -270,10 +271,9 @@ public FlowFileQueue createFlowFileQueue(final LoadBalanceStrategy loadBalanceSt public ReportingTaskNode createReportingTask(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, final boolean register, final String classloaderIsolationKey) { - if (type == null || id == null || bundleCoordinate == null) { Review Comment: Concurring with @joewitt, the previous implementation provides a much clearer error message than `requireNonNull`. Recommend reverting this change. ########## nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/logging/ParameterContextTaskLogObserver.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.logging; + +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.events.BulletinFactory; +import org.apache.nifi.reporting.Bulletin; +import org.apache.nifi.reporting.BulletinRepository; +import org.apache.nifi.reporting.ComponentType; +import org.apache.nifi.reporting.Severity; + +public class ParameterContextTaskLogObserver implements LogObserver { Review Comment: This should be renamed to `ParameterProviderLogObserver` to reflect the type of object being observed. ```suggestion public class ParameterProviderLogObserver implements LogObserver { ``` ########## nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java: ########## @@ -320,9 +321,55 @@ public ReportingTaskNode createReportingTask(final String type, final String id, @Override public ParameterProviderNode createParameterProvider(final String type, final String id, final BundleCoordinate bundleCoordinate, final Set<URL> additionalUrls, final boolean firstTimeAdded, - final boolean register) { + final boolean registerLogObserver) { + if (type == null || id == null || bundleCoordinate == null) { + throw new NullPointerException("Must supply type, id, and bundle coordinate in order to create Reporting Task. Provided arguments were type=" + type + ", id=" + id + + ", bundle coordinate = " + bundleCoordinate); + } + + // make sure the first reference to LogRepository happens outside of a NarCloseable so that we use the framework's ClassLoader + final LogRepository logRepository = LogRepositoryFactory.getRepository(id); + + final ParameterProviderNode taskNode; + try { + taskNode = new ComponentBuilder() + .identifier(id) + .type(type) + .bundleCoordinate(bundleCoordinate) + .statelessEngine(statelessEngine) + .additionalClassPathUrls(additionalUrls) + .flowManager(this) + .buildParameterProvider(); + } catch (ParameterProviderInstantiationException e) { + throw new RuntimeException("Could not create Parameter Context Task of type " + type + " with ID " + id, e); Review Comment: The wording should reference `Parameter Provider` instead of `Parameter Context`: ```suggestion throw new RuntimeException("Could not create Parameter Provider of type " + type + " with ID " + id, e); ``` -- 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-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
