markap14 commented on code in PR #11543:
URL: https://github.com/apache/nifi/pull/11543#discussion_r3786849735
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -2029,59 +2030,97 @@ public List<ConnectorMethod> getConnectorMethods() {
@Override
public String invokeConnectorMethod(final String methodName, final
Map<String, String> jsonArguments, final ProcessContext processContext) throws
InvocationFailedException {
- final ConfigurableComponent component = getComponent();
-
- try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(getExtensionManager(),
component.getClass(), getIdentifier())) {
- final Method implementationMethod =
discoverConnectorMethod(component.getClass(), methodName);
- final MethodArgument[] methodArguments =
getConnectorMethodArguments(methodName, implementationMethod, component);
- final List<Object> argumentValues = new ArrayList<>();
-
- for (final MethodArgument methodArgument : methodArguments) {
- if (ProcessContext.class.equals(methodArgument.type())) {
- continue;
- }
-
- final String jsonValue =
jsonArguments.get(methodArgument.name());
- if (jsonValue == null && methodArgument.required()) {
- throw new IllegalArgumentException("Cannot invoke
Connector Method '" + methodName + "' on " + this + " because the required
argument '"
- + methodArgument.name() + "' was not provided");
+ final boolean classpathDifferent =
isClasspathDifferent(processContext.getProperties());
+
+ if (classpathDifferent || isReloadAdditionalResourcesNecessary()) {
+ LOG.debug("Classpath reload required for Connector Method
invocation. Create temporary InstanceClassLoader for {}", this);
+ final ExtensionManager extensionManager = getExtensionManager();
+ final Bundle bundle =
extensionManager.getBundle(getBundleCoordinate());
+ final Set<URL> classpathUrls =
getAdditionalClasspathResources(processContext.getProperties().keySet(),
+ descriptor ->
processContext.getProperty(descriptor).getValue());
+ final String classloaderIsolationKey =
getClassLoaderIsolationKey(processContext);
+
+ final ClassLoader currentClassLoader =
Thread.currentThread().getContextClassLoader();
+ final InstanceClassLoader detectedClassLoader =
extensionManager.createInstanceClassLoader(getCanonicalClassName(),
getIdentifier(), bundle, classpathUrls, false,
+ classloaderIsolationKey);
+ try {
+
Thread.currentThread().setContextClassLoader(detectedClassLoader);
+ final Processor tempProcessor =
componentInstanceFactory.createProcessorInstance(this, detectedClassLoader);
+ try {
+ return invokeConnectorMethodOnComponent(tempProcessor,
methodName, jsonArguments, processContext);
+ } finally {
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class,
tempProcessor, processContext);
Review Comment:
Invoking `@OnRemoved` on the temporary instance is riskier here than it is
in `verifyConfiguration`, and I think it needs an explicit decision.
The temporary instance is initialized with the live component's identifier,
and it is handed the live `ProcessContext`. For controller services it is
worse: `StandardComponentInstanceFactory.createControllerServiceInstance`
passes `flowController.getStateManagerProvider().getStateManager(identifier)`,
so the temporary instance shares the live component's `StateManager`. An
`@OnRemoved` method that touches shared state or an external system will
therefore act on behalf of a component that has not been removed.
Two examples already in the tree:
* `GetSplunk.onRemoved` calls
`context.getStateManager().clear(Scope.CLUSTER)`.
* `ConsumeKinesis.onRemoved` deregisters an EFO consumer against AWS.
The existing precedent in `verifyConfiguration` is defensible because
verification is an explicit, user-initiated action and the component is
generally not running. Connector methods are different on both counts: they are
invoked implicitly by connector code (`DynamicAllowableValuesConnector` and
`KafkaToS3.getAvailableTopics` both call them from `fetchAllowableValues`), and
`ConnectorMethod.allowedStates` defaults to include `RUNNING`.
No shipped component currently has both a `@ConnectorMethod` and a
`dynamicallyModifiesClasspath` property, so this is not reachable today -- but
making that combination work is the point of NIFI-16201, so it becomes
reachable as soon as the feature is used as intended.
Worth considering whether the temporary instance should receive `@OnRemoved`
at all on this path, and if it should, documenting on
`ProcessorNode.invokeConnectorMethod` and
`ControllerServiceNode.invokeConnectorMethod` that a `@ConnectorMethod` may be
dispatched to a throwaway instance whose lifecycle methods have not run and
whose `@OnRemoved` will be invoked afterwards. Neither declaration currently
has any Javadoc.
Same comment applies to the corresponding line in
`StandardControllerServiceNode`.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -2029,59 +2030,97 @@ public List<ConnectorMethod> getConnectorMethods() {
@Override
public String invokeConnectorMethod(final String methodName, final
Map<String, String> jsonArguments, final ProcessContext processContext) throws
InvocationFailedException {
- final ConfigurableComponent component = getComponent();
-
- try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(getExtensionManager(),
component.getClass(), getIdentifier())) {
- final Method implementationMethod =
discoverConnectorMethod(component.getClass(), methodName);
- final MethodArgument[] methodArguments =
getConnectorMethodArguments(methodName, implementationMethod, component);
- final List<Object> argumentValues = new ArrayList<>();
-
- for (final MethodArgument methodArgument : methodArguments) {
- if (ProcessContext.class.equals(methodArgument.type())) {
- continue;
- }
-
- final String jsonValue =
jsonArguments.get(methodArgument.name());
- if (jsonValue == null && methodArgument.required()) {
- throw new IllegalArgumentException("Cannot invoke
Connector Method '" + methodName + "' on " + this + " because the required
argument '"
- + methodArgument.name() + "' was not provided");
+ final boolean classpathDifferent =
isClasspathDifferent(processContext.getProperties());
+
+ if (classpathDifferent || isReloadAdditionalResourcesNecessary()) {
+ LOG.debug("Classpath reload required for Connector Method
invocation. Create temporary InstanceClassLoader for {}", this);
+ final ExtensionManager extensionManager = getExtensionManager();
+ final Bundle bundle =
extensionManager.getBundle(getBundleCoordinate());
+ final Set<URL> classpathUrls =
getAdditionalClasspathResources(processContext.getProperties().keySet(),
+ descriptor ->
processContext.getProperty(descriptor).getValue());
+ final String classloaderIsolationKey =
getClassLoaderIsolationKey(processContext);
+
+ final ClassLoader currentClassLoader =
Thread.currentThread().getContextClassLoader();
+ final InstanceClassLoader detectedClassLoader =
extensionManager.createInstanceClassLoader(getCanonicalClassName(),
getIdentifier(), bundle, classpathUrls, false,
+ classloaderIsolationKey);
+ try {
+
Thread.currentThread().setContextClassLoader(detectedClassLoader);
+ final Processor tempProcessor =
componentInstanceFactory.createProcessorInstance(this, detectedClassLoader);
+ try {
+ return invokeConnectorMethodOnComponent(tempProcessor,
methodName, jsonArguments, processContext);
+ } finally {
+
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnRemoved.class,
tempProcessor, processContext);
}
+ } catch (final ProcessorInstantiationException e) {
+ throw new InvocationFailedException("Failed to create
temporary Processor instance for Connector Method '" + methodName + "' on " +
this, e);
+ } finally {
+
Thread.currentThread().setContextClassLoader(currentClassLoader);
- if (jsonValue == null) {
- argumentValues.add(null);
- } else {
- try {
- final Object argumentValue =
OBJECT_MAPPER.readValue(jsonValue, methodArgument.type());
- argumentValues.add(argumentValue);
- } catch (final JsonProcessingException e) {
- throw new InvocationFailedException("Failed to
deserialize argument '" + methodArgument.name() + "' as type " +
methodArgument.type().getName() +
- " for Connector
Method '" + methodName + "' on " + this, e);
- }
+ try {
+ detectedClassLoader.close();
Review Comment:
Closing the temporary ClassLoader can permanently disable shared-ClassLoader
reuse for a ClassLoader isolation key.
In `StandardExtensionDiscoveringManager.createInstanceClassLoader`, when no
`SharedInstanceClassLoader` exists yet for the `BaseClassLoaderKey`, a new one
is created with a reference count of 1 and stored with
`sharedBaseClassloaders.putIfAbsent(...)`. `InstanceClassLoader.close()`
cascades to the parent `SharedInstanceClassLoader.close()`, which decrements
the count to 0 and sets `closed = true`. Nothing ever removes the entry from
`sharedBaseClassloaders`.
So if a temporary invocation happens to be the first to create the shared
base ClassLoader for a key, the map is left holding a closed instance. Every
later instance for that key then gets `incrementReferenceCount() == false`,
clones the full NAR and ancestor-NAR URL set into its own
`SharedInstanceClassLoader`, and `putIfAbsent` silently fails to replace the
dead entry -- so sharing is broken for that key for the lifetime of the JVM,
and memory grows with the number of instances.
This is reachable when the live component was instantiated with a null
isolation key (the `allowsSharedClassloader && classloaderIsolationKey == null`
branch creates a plain `InstanceClassLoader` and registers nothing) and a later
connector method invocation resolves a non-null key.
The hazard exists in `verifyConfiguration` today, so I am not asking this PR
to fix `StandardExtensionDiscoveringManager`. But connector methods are invoked
far more often and far less deliberately than verification, so it is worth
deciding whether that is acceptable here or whether the map should evict closed
entries.
Minor, while this line is being touched: `verifyConfiguration` manages the
same object with try-with-resources. Doing the same here would remove the
hand-rolled `close()` and its `IOException` handling.
--
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]
