This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 7edf0d4e2a NIFI-12693: Moved notification of python process that a
Processor was removed to a background (virtual) thread. Also noted in testing
that in one instance a Python Processor never became valid because it had
cached property descriptors before the processor was fully initialized, so
updated code to ensure that we do not cache values before initialization is
completed.
7edf0d4e2a is described below
commit 7edf0d4e2aee21949bc3ef755ca4b27fa324d20f
Author: Mark Payne <[email protected]>
AuthorDate: Tue Jan 30 10:58:22 2024 -0500
NIFI-12693: Moved notification of python process that a Processor was
removed to a background (virtual) thread. Also noted in testing that in one
instance a Python Processor never became
valid because it had cached property descriptors before the processor was
fully initialized, so updated code to ensure that we do not cache values before
initialization is completed.
This closes #8315
Signed-off-by: David Handermann <[email protected]>
---
.../apache/nifi/groups/StandardProcessGroup.java | 21 ++++++------
.../nifi/web/dao/impl/StandardProcessorDAO.java | 2 +-
.../org/apache/nifi/py4j/StandardPythonBridge.java | 37 ++++++++++++++--------
.../python/processor/PythonProcessorProxy.java | 8 ++---
4 files changed, 38 insertions(+), 30 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index 78579458e3..e9db230087 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -1263,6 +1263,14 @@ public final class StandardProcessGroup implements
ProcessGroup {
}
}
+ // Remove connections prior to removing the Processor. If there is
any failure in removing the Processor or the associated cleanup,
+ // we can handle that. However, we could have many potential
issues if Connections exist whose source or destination does not exist.
+ // must copy to avoid a concurrent modification
+ final List<Connection> copy = new
ArrayList<>(processor.getConnections());
+ for (final Connection conn : copy) {
+ removeConnection(conn);
+ }
+
processors.remove(id);
onComponentModified();
@@ -1274,18 +1282,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
logRepository.removeAllObservers();
}
- scheduler.submitFrameworkTask(new Runnable() {
- @Override
- public void run() {
-
stateManagerProvider.onComponentRemoved(processor.getIdentifier());
- }
- });
-
- // must copy to avoid a concurrent modification
- final Set<Connection> copy = new
HashSet<>(processor.getConnections());
- for (final Connection conn : copy) {
- removeConnection(conn);
- }
+ scheduler.submitFrameworkTask(() ->
stateManagerProvider.onComponentRemoved(processor.getIdentifier()));
removed = true;
LOG.info("{} removed from flow", processor);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardProcessorDAO.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardProcessorDAO.java
index c4198fe282..552b82fdeb 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardProcessorDAO.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardProcessorDAO.java
@@ -589,7 +589,7 @@ public class StandardProcessorDAO extends ComponentDAO
implements ProcessorDAO {
try {
// attempt remove the processor
processor.getProcessGroup().removeProcessor(processor);
- } catch (ComponentLifeCycleException plce) {
+ } catch (final ComponentLifeCycleException plce) {
throw new NiFiCoreException(plce.getMessage(), plce);
}
}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
index eb568dfdad..e3f735bb01 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
@@ -38,11 +38,11 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -162,18 +162,29 @@ public class StandardPythonBridge implements PythonBridge
{
return;
}
- // Find the Python Process that has the Processor, if any, and
remove it.
- // If there are no additional Processors in the Python Process,
remove it from our list and shut down the process.
- final Iterator<PythonProcess> processItr = processes.iterator();
// Use iterator so we can call remove()
- while (processItr.hasNext()) {
- final PythonProcess process = processItr.next();
- final boolean removed = process.removeProcessor(identifier);
- if (removed && process.getProcessorCount() == 0) {
- processItr.remove();
- process.shutdown();
- break;
+ Thread.ofVirtual().name("Remove Python Processor " +
identifier).start(() -> {
+ PythonProcess toRemove = null;
+
+ try {
+ // Find the Python Process that has the Processor, if any,
and remove it.
+ // If there are no additional Processors in the Python
Process, remove it from our list and shut down the process.
+ // Use iterator so we can call remove()
+ for (final PythonProcess process : processes) {
+ final boolean removed =
process.removeProcessor(identifier);
+ if (removed && process.getProcessorCount() == 0) {
+ toRemove = process;
+ break;
+ }
+ }
+
+ if (toRemove != null) {
+ processes.remove(toRemove);
+ toRemove.shutdown();
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to trigger removal of Python
Processor with ID {}", identifier, e);
}
- }
+ });
processorCountByType.merge(extensionId, -1, Integer::sum);
} else {
@@ -198,7 +209,7 @@ public class StandardPythonBridge implements PythonBridge {
// isolation (which is the case when Extension Manager creates a temp
component), or if an existing process
// consists only of processors that don't prefer isolation. I.e., we
don't want to collocate two Processors if
// they both prefer isolation.
- final List<PythonProcess> processesForType =
processesByProcessorType.computeIfAbsent(extensionId, key -> new ArrayList<>());
+ final List<PythonProcess> processesForType =
processesByProcessorType.computeIfAbsent(extensionId, key -> new
CopyOnWriteArrayList<>());
for (final PythonProcess pythonProcess : processesForType) {
if (!preferIsolatedProcess ||
!pythonProcess.containsIsolatedProcessor()) {
logger.debug("Using {} to create Processor of type {}",
pythonProcess, extensionId.type());
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/PythonProcessorProxy.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/PythonProcessorProxy.java
index 9ee4fcb212..a20ad583c2 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/PythonProcessorProxy.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/python/processor/PythonProcessorProxy.java
@@ -126,7 +126,7 @@ public abstract class PythonProcessorProxy extends
AbstractProcessor implements
return this.cachedPropertyDescriptors;
}
- if (bridge == null) {
+ if (getState() != LoadState.FINISHED_LOADING) {
return Collections.emptyList();
}
@@ -202,7 +202,7 @@ public abstract class PythonProcessorProxy extends
AbstractProcessor implements
return cachedDynamicDescriptors.get(propertyDescriptorName);
}
- if (bridge == null) {
+ if (getState() != LoadState.FINISHED_LOADING) {
return null;
}
@@ -221,7 +221,7 @@ public abstract class PythonProcessorProxy extends
AbstractProcessor implements
return supportsDynamicProperties;
}
- if (bridge == null) {
+ if (getState() != LoadState.FINISHED_LOADING) {
return false;
}
@@ -266,7 +266,7 @@ public abstract class PythonProcessorProxy extends
AbstractProcessor implements
}
private Set<Relationship> fetchRelationshipsFromPythonProcessor() {
- if (bridge == null) {
+ if (getState() != LoadState.FINISHED_LOADING) {
return Collections.emptySet();
}
