NIFI-13997 Limit Python Processor loading to requested Processor Class (#9568)

Signed-off-by: David Handermann <[email protected]>

Author: fgerlits

.github/workflows/system-tests.yml

@@ -61,6 +61,7 @@ env:
     -pl :nifi-python-framework
     -pl :nifi-python-extension-api
     -pl :nifi-python-test-extensions
+    -pl :nifi-py4j-integration-tests
     -pl nifi-system-tests/nifi-system-test-suite
     -pl nifi-system-tests/nifi-stateless-system-test-suite
 

nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java

@@ -613,6 +613,16 @@ public void testCreateNothing() {
         runner.assertTransferCount("success", 0);
     }
 
+    @Test
+    public void testMultipleProcessorsInAPackage() {
+        // This processor is in a package with another processor, which has additional dependency requirements
+        final TestRunner runner = createProcessor("CreateHttpRequest");
+        waitForValid(runner);
+        runner.run();
+
+        runner.assertTransferCount("success", 1);
+    }
+
     @Test
     public void testStateManagerSetState() {
         final TestRunner runner = createStateManagerTesterProcessor("setState");

nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/multi-processor/CreateHttpRequest.py

@@ -0,0 +1,34 @@
+# 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.
+
+import requests
+from nifiapi.flowfilesource import FlowFileSource, FlowFileSourceResult
+
+class CreateHttpRequest(FlowFileSource):
+    class Java:
+        implements = ['org.apache.nifi.python.processor.FlowFileSource']
+
+    class ProcessorDetails:
+        description = "Test processor which creates an http request"
+        version = '0.0.1-SNAPSHOT'
+        tags = ['test', 'http', 'requests']
+        dependencies = ['requests==2.32.3']
+
+    def __init__(self, jvm):
+        pass
+
+    def create(self, context):
+        req = requests.Request('GET', 'https://wikipedia.org')
+        return FlowFileSourceResult('success', contents=str(req))

nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/multi-processor/ProtobufVersion.py

@@ -0,0 +1,34 @@
+# 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.
+
+from google.protobuf.runtime_version import OSS_MAJOR, OSS_MINOR, OSS_PATCH
+from nifiapi.flowfilesource import FlowFileSource, FlowFileSourceResult
+
+class ProtobufVersion(FlowFileSource):
+    class Java:
+        implements = ['org.apache.nifi.python.processor.FlowFileSource']
+
+    class ProcessorDetails:
+        description = "Test processor which outputs the version of the protobuf library"
+        version = '0.0.1-SNAPSHOT'
+        tags = ['test', 'protobuf', 'version']
+        dependencies = ['protobuf==5.29.1']
+
+    def __init__(self, jvm):
+        pass
+
+    def create(self, context):
+        version = f"{OSS_MAJOR}.{OSS_MINOR}.{OSS_PATCH}"
+        return FlowFileSourceResult('success', contents=version)

nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/multi-processor/__init__.py

@@ -0,0 +1,14 @@
+# 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.

nifi-extension-bundles/nifi-py4j-extension-bundle/nifi-python-test-extensions/src/main/resources/extensions/multi-processor/requirements.txt

@@ -0,0 +1,16 @@
+# 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.
+
+charset-normalizer==3.4.0

nifi-framework-bundle/nifi-framework-extensions/nifi-py4j-framework-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py

@@ -319,9 +319,7 @@ def import_external_dependencies(self, processor_details, work_dir):
     def __load_extension_module(self, file, local_dependencies):
         # If there are any local dependencies (i.e., other python files in the same directory), load those modules first
         if local_dependencies:
-            to_load = [dep for dep in local_dependencies]
-            if file in to_load:
-                to_load.remove(file)
+            to_load = [dep for dep in local_dependencies if dep != file and not self.__is_processor_module(dep)]
 
             # There is almost certainly a better way to do this. But we need to load all modules that are 'local dependencies'. I.e., all
             # modules in the same directory/package. But Python does not appear to give us a simple way to do this. We could have a situation in which
@@ -378,6 +376,10 @@ def __load_extension_module(self, file, local_dependencies):
         return None
 
 
+    def __is_processor_module(self, module_file):
+        return len(ProcessorInspection.get_processor_class_nodes(module_file)) > 0
+
+
     def __is_processor_class(self, potential_processor_class):
         # Go through all members of the given class and see if it has an inner class named Java
         for name, member in inspect.getmembers(potential_processor_class):