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 b523698 NIFI-8212: Refactored StandardExtensionDiscoveringManager to
avoid using ServiceLoader
b523698 is described below
commit b523698534679938865a3ac401d901e8db411225
Author: Mark Payne <[email protected]>
AuthorDate: Thu Jan 7 09:31:45 2021 -0500
NIFI-8212: Refactored StandardExtensionDiscoveringManager to avoid using
ServiceLoader
Instead, it will look at the ServiceLoader file and read the names of the
classes but avoid instantiating all of the objects or loading the classes into
memory.
- Updated Doc Generation so that if the documentation for a given NAR
already exists, it doesn't delete it and re-generate it. This was necessary
because we are no longer instantiating an instance of each component and
instead lazily creating the components as necessary.
- Removed stateless version of extension registry because it's no longer
necessary
This closes #4852
Signed-off-by: David Handermann <[email protected]>
---
.../apache/nifi/documentation/DocGenerator.java | 38 ++-
.../html/HtmlDocumentationWriter.java | 6 +-
.../scheduling/ConnectableProcessContext.java | 26 +-
.../controller/service/ServiceStateTransition.java | 43 ++-
.../service/StandardControllerServiceNode.java | 28 +-
.../service/StandardControllerServiceProvider.java | 21 +-
.../apache/nifi/groups/StandardProcessGroup.java | 33 ++-
.../nifi/processor/StandardProcessContext.java | 11 +-
.../controller/service/ControllerServiceNode.java | 20 +-
.../apache/nifi/controller/ExtensionBuilder.java | 92 ++++++-
.../org/apache/nifi/controller/FlowController.java | 16 +-
.../nifi/controller/StandardFlowSnippet.java | 14 +-
.../nifi/provenance/ComponentIdentifierLookup.java | 11 +-
.../DirectInjectionExtensionManager.java | 4 +-
.../org/apache/nifi/nar/StandardNarLoader.java | 10 +-
.../org/apache/nifi/nar/ExtensionDefinition.java | 88 ++++++
.../java/org/apache/nifi/nar/ExtensionManager.java | 15 +-
.../nar/StandardExtensionDiscoveringManager.java | 305 ++++++++++++---------
.../main/java/org/apache/nifi/nar/NarUnpacker.java | 41 +--
.../main/java/org/apache/nifi/StatelessNiFi.java | 2 +-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 10 +-
.../nifi/web/controller/ControllerFacade.java | 10 +-
.../stateless/bootstrap/StatelessBootstrap.java | 17 +-
.../extensions/FileSystemExtensionRepository.java | 6 +-
.../stateless/bootstrap/ExtensionDiscovery.java | 5 +-
.../stateless/engine/StandardStatelessEngine.java | 39 ++-
.../nifi/stateless/engine/StatelessEngine.java | 2 +-
.../stateless/engine/StatelessFlowManager.java | 2 +-
.../flow/StandardStatelessDataflowFactory.java | 27 +-
.../nifi/stateless/flow/StandardStatelessFlow.java | 46 ++--
.../nifi/processors/standard/TestListenTCP.java | 19 +-
.../apache/nifi/stateless/StatelessSystemIT.java | 2 +-
.../classloader/InstanceClassLoaderIT.java | 100 +++++++
.../nifi/processors/tests/system/PassThrough.java | 51 ++++
.../PassThroughRequiresInstanceClassLoading.java | 53 ++++
.../services/org.apache.nifi.processor.Processor | 2 +
.../clustering/JoinClusterWithDifferentFlow.java | 17 +-
37 files changed, 892 insertions(+), 340 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/DocGenerator.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/DocGenerator.java
index aa2dbe0..3f69c58 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/DocGenerator.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/DocGenerator.java
@@ -22,6 +22,7 @@ import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.documentation.html.HtmlDocumentationWriter;
import org.apache.nifi.documentation.html.HtmlProcessorDocumentationWriter;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.ExtensionMapping;
import org.apache.nifi.processor.Processor;
@@ -69,25 +70,34 @@ public class DocGenerator {
/**
* Documents a type of configurable component.
*
- * @param extensionClasses types of a configurable component
+ * @param extensionDefinitions definitions of the extensions to document
* @param explodedNiFiDocsDir base directory of component documentation
*/
- public static void documentConfigurableComponent(final Set<Class>
extensionClasses, final File explodedNiFiDocsDir, final ExtensionManager
extensionManager) {
- for (final Class<?> extensionClass : extensionClasses) {
- if (ConfigurableComponent.class.isAssignableFrom(extensionClass)) {
- final String extensionClassName =
extensionClass.getCanonicalName();
-
- final Bundle bundle =
extensionManager.getBundle(extensionClass.getClassLoader());
- if (bundle == null) {
- logger.warn("No coordinate found for {}, skipping...", new
Object[] {extensionClassName});
- continue;
- }
- final BundleCoordinate coordinate =
bundle.getBundleDetails().getCoordinate();
+ public static void documentConfigurableComponent(final
Set<ExtensionDefinition> extensionDefinitions, final File explodedNiFiDocsDir,
final ExtensionManager extensionManager) {
+ for (final ExtensionDefinition extensionDefinition :
extensionDefinitions) {
+ final Bundle bundle = extensionDefinition.getBundle();
+ if (bundle == null) {
+ logger.warn("Cannot document extension {} because it has no
bundle associated with it", extensionDefinition);
+ continue;
+ }
+
+ final BundleCoordinate coordinate =
bundle.getBundleDetails().getCoordinate();
+
+ final String extensionClassName =
extensionDefinition.getImplementationClassName();
+ final String path = coordinate.getGroup() + "/" +
coordinate.getId() + "/" + coordinate.getVersion() + "/" + extensionClassName;
+ final File componentDirectory = new File(explodedNiFiDocsDir,
path);
+ final File indexHtml = new File(componentDirectory, "index.html");
+ if (indexHtml.exists()) {
+ // index.html already exists, no need to unpack the docs again.
+ logger.debug("Found existing documentation file {}. Will not
generate documentation for {}", indexHtml.getAbsolutePath(),
extensionClassName);
+ continue;
+ }
- final String path = coordinate.getGroup() + "/" +
coordinate.getId() + "/" + coordinate.getVersion() + "/" + extensionClassName;
- final File componentDirectory = new File(explodedNiFiDocsDir,
path);
+ final Class<?> extensionType =
extensionDefinition.getExtensionType();
+ if (ConfigurableComponent.class.isAssignableFrom(extensionType)) {
componentDirectory.mkdirs();
+ final Class<?> extensionClass =
extensionManager.getClass(extensionDefinition);
final Class<? extends ConfigurableComponent> componentClass =
extensionClass.asSubclass(ConfigurableComponent.class);
try {
logger.debug("Documenting: " + componentClass);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/html/HtmlDocumentationWriter.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/html/HtmlDocumentationWriter.java
index 938ee9e..ce8f952 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/html/HtmlDocumentationWriter.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-documentation/src/main/java/org/apache/nifi/documentation/html/HtmlDocumentationWriter.java
@@ -37,6 +37,7 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.documentation.DocumentationWriter;
import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.util.StringUtils;
import org.slf4j.Logger;
@@ -901,11 +902,12 @@ public class HtmlDocumentationWriter implements
DocumentationWriter {
final List<Class<? extends ControllerService>> implementations = new
ArrayList<>();
// first get all ControllerService implementations
- final Set<Class> controllerServices =
extensionManager.getExtensions(ControllerService.class);
+ final Set<ExtensionDefinition> controllerServices =
extensionManager.getExtensions(ControllerService.class);
// then iterate over all controller services looking for any that is a
child of the parent
// ControllerService API that was passed in as a parameter
- for (final Class<? extends ControllerService> controllerServiceClass :
controllerServices) {
+ for (final ExtensionDefinition extensionDefinition :
controllerServices) {
+ final Class controllerServiceClass =
extensionManager.getClass(extensionDefinition);
if (parent.isAssignableFrom(controllerServiceClass)) {
implementations.add(controllerServiceClass);
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/scheduling/ConnectableProcessContext.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/scheduling/ConnectableProcessContext.java
index 629c988..e02a395 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/scheduling/ConnectableProcessContext.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/scheduling/ConnectableProcessContext.java
@@ -16,14 +16,6 @@
*/
package org.apache.nifi.controller.repository.scheduling;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.state.StateManager;
@@ -41,18 +33,26 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.util.Connectables;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
/**
* This class is essentially an empty shell for {@link Connectable}s that are
not Processors
*/
public class ConnectableProcessContext implements ProcessContext {
private final Connectable connectable;
- private final PropertyEncryptor encryptor;
+ private final PropertyEncryptor propertyEncryptor;
private final StateManager stateManager;
- public ConnectableProcessContext(final Connectable connectable, final
PropertyEncryptor encryptor, final StateManager stateManager) {
+ public ConnectableProcessContext(final Connectable connectable, final
PropertyEncryptor propertyEncryptor, final StateManager stateManager) {
this.connectable = connectable;
- this.encryptor = encryptor;
+ this.propertyEncryptor = propertyEncryptor;
this.stateManager = stateManager;
}
@@ -212,12 +212,12 @@ public class ConnectableProcessContext implements
ProcessContext {
@Override
public String decrypt(String encrypted) {
- return encryptor.decrypt(encrypted);
+ return propertyEncryptor.decrypt(encrypted);
}
@Override
public String encrypt(String unencrypted) {
- return encryptor.encrypt(unencrypted);
+ return propertyEncryptor.encrypt(unencrypted);
}
@Override
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
index a2cd537..2971764 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
@@ -17,14 +17,22 @@
package org.apache.nifi.controller.service;
+import org.apache.nifi.controller.ComponentNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ServiceStateTransition {
+ private static final Logger logger =
LoggerFactory.getLogger(ServiceStateTransition.class);
private ControllerServiceState state = ControllerServiceState.DISABLED;
private final List<CompletableFuture<?>> enabledFutures = new
ArrayList<>();
private final List<CompletableFuture<?>> disabledFutures = new
ArrayList<>();
@@ -32,6 +40,7 @@ public class ServiceStateTransition {
private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Lock writeLock = rwLock.writeLock();
private final Lock readLock = rwLock.readLock();
+ private final Condition stateChangeCondition = writeLock.newCondition();
public boolean transitionToEnabling(final ControllerServiceState
expectedState, final CompletableFuture<?> enabledFuture) {
writeLock.lock();
@@ -41,6 +50,7 @@ public class ServiceStateTransition {
}
state = ControllerServiceState.ENABLING;
+ stateChangeCondition.signalAll();
enabledFutures.add(enabledFuture);
return true;
} finally {
@@ -48,7 +58,7 @@ public class ServiceStateTransition {
}
}
- public boolean enable() {
+ public boolean enable(final ControllerServiceReference
controllerServiceReference) {
writeLock.lock();
try {
if (state != ControllerServiceState.ENABLING) {
@@ -58,6 +68,14 @@ public class ServiceStateTransition {
state = ControllerServiceState.ENABLED;
enabledFutures.forEach(future -> future.complete(null));
+
+ final List<ComponentNode> referencingComponents =
controllerServiceReference.findRecursiveReferences(ComponentNode.class);
+ for (final ComponentNode component : referencingComponents) {
+ component.performValidation();
+ }
+
+ stateChangeCondition.signalAll();
+
return true;
} finally {
writeLock.unlock();
@@ -72,6 +90,7 @@ public class ServiceStateTransition {
}
state = ControllerServiceState.DISABLING;
+ stateChangeCondition.signalAll();
disabledFutures.add(disabledFuture);
return true;
} finally {
@@ -83,6 +102,7 @@ public class ServiceStateTransition {
writeLock.lock();
try {
state = ControllerServiceState.DISABLED;
+ stateChangeCondition.signalAll();
disabledFutures.forEach(future -> future.complete(null));
} finally {
writeLock.unlock();
@@ -97,4 +117,25 @@ public class ServiceStateTransition {
readLock.unlock();
}
}
+
+ public boolean awaitState(final ControllerServiceState desiredState, final
long timePeriod, final TimeUnit timeUnit) throws InterruptedException {
+ Objects.requireNonNull(timeUnit);
+ final long timeout = System.currentTimeMillis() +
timeUnit.toMillis(timePeriod);
+
+ writeLock.lock();
+ try {
+ while (desiredState != state) {
+ final long millisLeft = timeout - System.currentTimeMillis();
+ if (millisLeft <= 0) {
+ return false;
+ }
+
+ stateChangeCondition.await(millisLeft, TimeUnit.MILLISECONDS);
+ }
+
+ return true;
+ } finally {
+ writeLock.unlock();
+ }
+ }
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index c9b632b..6167a5e 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -28,6 +28,7 @@ import org.apache.nifi.authorization.resource.ResourceType;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.validation.ValidationState;
import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.components.validation.ValidationTrigger;
import org.apache.nifi.controller.AbstractComponentNode;
@@ -367,6 +368,19 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
return this.active.get();
}
+ public boolean awaitEnabled(final long timePeriod, final TimeUnit
timeUnit) throws InterruptedException {
+ LOG.debug("Waiting up to {} {} for {} to be enabled", timePeriod,
timeUnit, this);
+ final boolean enabled =
stateTransition.awaitState(ControllerServiceState.ENABLED, timePeriod,
timeUnit);
+
+ if (enabled) {
+ LOG.debug("{} is enabled", this);
+ } else {
+ LOG.debug("After {} {}, {} is NOT enabled", timePeriod, timeUnit,
this);
+ }
+
+ return enabled;
+ }
+
@Override
public boolean isValidationNecessary() {
switch (getState()) {
@@ -427,7 +441,10 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
final ValidationStatus validationStatus =
getValidationStatus();
if (validationStatus != ValidationStatus.VALID) {
- LOG.debug("Cannot enable {} because it is not
currently valid. (Validation State is {}). Will try again in 1 second",
StandardControllerServiceNode.this, getValidationState());
+ final ValidationState validationState =
getValidationState();
+ LOG.debug("Cannot enable {} because it is not
currently valid. (Validation State is {}: {}). Will try again in 1 second",
+ StandardControllerServiceNode.this,
validationState, validationState.getValidationErrors());
+
scheduler.schedule(this, 1, TimeUnit.SECONDS);
future.complete(null);
return;
@@ -440,9 +457,8 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
boolean shouldEnable;
synchronized (active) {
- shouldEnable = active.get() &&
stateTransition.enable(); // Transitioning the state to ENABLED will complete
our future.
+ shouldEnable = active.get() &&
stateTransition.enable(getReferences()); // Transitioning the state to ENABLED
will complete our future.
}
- validateReferences();
if (!shouldEnable) {
LOG.info("Disabling service {} after it has been
enabled due to disable action being initiated.", service);
@@ -481,12 +497,6 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
return future;
}
- private void validateReferences() {
- final List<ComponentNode> referencingComponents =
getReferences().findRecursiveReferences(ComponentNode.class);
- for (final ComponentNode component : referencingComponents) {
- component.performValidation();
- }
- }
/**
* Will atomically disable this service by invoking its @OnDisabled
operation.
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 4cdd49c..4e17aab 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -36,7 +36,6 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -292,12 +291,10 @@ public class StandardControllerServiceProvider implements
ControllerServiceProvi
return CompletableFuture.completedFuture(null);
}
- final Map<ControllerServiceNode, Future<Void>> futures = new
HashMap<>();
-
- for (ControllerServiceNode depNode :
serviceNode.getRequiredControllerServices()) {
+ final List<ControllerServiceNode> dependentServices =
serviceNode.getRequiredControllerServices();
+ for (final ControllerServiceNode depNode : dependentServices) {
if (!depNode.isActive()) {
logger.debug("Before enabling {}, will enable dependent
Controller Service {}", serviceNode, depNode);
- futures.put(depNode,
this.enableControllerServiceAndDependencies(depNode));
}
}
@@ -305,13 +302,15 @@ public class StandardControllerServiceProvider implements
ControllerServiceProvi
logger.debug("All dependent services for {} have now begun
enabling. Will wait for them to complete", serviceNode);
}
- for (final Map.Entry<ControllerServiceNode, Future<Void>> entry :
futures.entrySet()) {
- final ControllerServiceNode dependentService = entry.getKey();
- final Future<Void> future = entry.getValue();
-
+ for (final ControllerServiceNode dependentService : dependentServices)
{
try {
- future.get(30, TimeUnit.SECONDS);
- logger.debug("Successfully enabled dependent service {};
service state = {}", dependentService, dependentService.getState());
+ final boolean enabled = dependentService.awaitEnabled(30,
TimeUnit.SECONDS);
+
+ if (enabled) {
+ logger.debug("Successfully enabled dependent service {};
service state = {}", dependentService, dependentService.getState());
+ } else {
+ logger.debug("After 30 seconds, {} is still not enabled.
Will continue attempting to enable additional Controller Services",
dependentService);
+ }
} catch (final Exception e) {
logger.error("Failed to enable service {}, so may be unable to
enable {}", dependentService, serviceNode, e);
// Nothing we can really do. Will attempt to enable this
service anyway.
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 965a729..4191ff3 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
@@ -5198,12 +5198,20 @@ public final class StandardProcessGroup implements
ProcessGroup {
final String processorToAddClass = processorToAdd.getType();
final BundleCoordinate processorToAddCoordinate =
toCoordinate(processorToAdd.getBundle());
- final List<org.apache.nifi.bundle.Bundle> possibleBundles =
extensionManager.getBundles(processorToAddClass);
- final boolean bundleExists = possibleBundles.stream()
- .anyMatch(b ->
processorToAddCoordinate.equals(b.getBundleDetails().getCoordinate()));
-
- if (!bundleExists && possibleBundles.size() != 1) {
- throw new IllegalArgumentException("Unknown bundle " +
processorToAddCoordinate.toString() + " for processor type " +
processorToAddClass);
+ // Get the exact bundle requested, if it exists.
+ final Bundle bundle = processorToAdd.getBundle();
+ final BundleCoordinate coordinate = new
BundleCoordinate(bundle.getGroup(), bundle.getArtifact(), bundle.getVersion());
+ final org.apache.nifi.bundle.Bundle resolved =
extensionManager.getBundle(coordinate);
+
+ if (resolved == null) {
+ // Could not resolve the bundle explicitly. Check for
possible bundles.
+ final List<org.apache.nifi.bundle.Bundle> possibleBundles
= extensionManager.getBundles(processorToAddClass);
+ final boolean bundleExists = possibleBundles.stream()
+ .anyMatch(b ->
processorToAddCoordinate.equals(b.getBundleDetails().getCoordinate()));
+
+ if (!bundleExists && possibleBundles.size() != 1) {
+ throw new IllegalArgumentException("Unknown bundle " +
processorToAddCoordinate.toString() + " for processor type " +
processorToAddClass);
+ }
}
}
@@ -5219,12 +5227,15 @@ public final class StandardProcessGroup implements
ProcessGroup {
final String serviceToAddClass = serviceToAdd.getType();
final BundleCoordinate serviceToAddCoordinate =
toCoordinate(serviceToAdd.getBundle());
- final List<org.apache.nifi.bundle.Bundle> possibleBundles =
extensionManager.getBundles(serviceToAddClass);
- final boolean bundleExists = possibleBundles.stream()
- .anyMatch(b ->
serviceToAddCoordinate.equals(b.getBundleDetails().getCoordinate()));
+ final org.apache.nifi.bundle.Bundle resolved =
extensionManager.getBundle(serviceToAddCoordinate);
+ if (resolved == null) {
+ final List<org.apache.nifi.bundle.Bundle> possibleBundles
= extensionManager.getBundles(serviceToAddClass);
+ final boolean bundleExists = possibleBundles.stream()
+ .anyMatch(b ->
serviceToAddCoordinate.equals(b.getBundleDetails().getCoordinate()));
- if (!bundleExists && possibleBundles.size() != 1) {
- throw new IllegalArgumentException("Unknown bundle " +
serviceToAddCoordinate.toString() + " for service type " + serviceToAddClass);
+ if (!bundleExists && possibleBundles.size() != 1) {
+ throw new IllegalArgumentException("Unknown bundle " +
serviceToAddCoordinate.toString() + " for service type " + serviceToAddClass);
+ }
}
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
index ec5456a..f9e14f7 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/processor/StandardProcessContext.java
@@ -49,17 +49,18 @@ public class StandardProcessContext implements
ProcessContext, ControllerService
private final ProcessorNode procNode;
private final ControllerServiceProvider controllerServiceProvider;
private final Map<PropertyDescriptor, PreparedQuery> preparedQueries;
- private final PropertyEncryptor encryptor;
+ private final PropertyEncryptor propertyEncryptor;
private final StateManager stateManager;
private final TaskTermination taskTermination;
private final NodeTypeProvider nodeTypeProvider;
private final Map<PropertyDescriptor, String> properties;
- public StandardProcessContext(final ProcessorNode processorNode, final
ControllerServiceProvider controllerServiceProvider, final PropertyEncryptor
encryptor,
+
+ public StandardProcessContext(final ProcessorNode processorNode, final
ControllerServiceProvider controllerServiceProvider, final PropertyEncryptor
propertyEncryptor,
final StateManager stateManager, final
TaskTermination taskTermination, final NodeTypeProvider nodeTypeProvider) {
this.procNode = processorNode;
this.controllerServiceProvider = controllerServiceProvider;
- this.encryptor = encryptor;
+ this.propertyEncryptor = propertyEncryptor;
this.stateManager = stateManager;
this.taskTermination = taskTermination;
this.nodeTypeProvider = nodeTypeProvider;
@@ -178,13 +179,13 @@ public class StandardProcessContext implements
ProcessContext, ControllerService
@Override
public String encrypt(final String unencrypted) {
verifyTaskActive();
- return encryptor.encrypt(unencrypted);
+ return propertyEncryptor.encrypt(unencrypted);
}
@Override
public String decrypt(final String encrypted) {
verifyTaskActive();
- return encryptor.decrypt(encrypted);
+ return propertyEncryptor.decrypt(encrypted);
}
@Override
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
index 78318aa..5d542df 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/service/ControllerServiceNode.java
@@ -16,11 +16,6 @@
*/
package org.apache.nifi.controller.service;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.VersionedComponent;
import org.apache.nifi.controller.ComponentNode;
@@ -28,6 +23,12 @@ import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.LoggableComponent;
import org.apache.nifi.groups.ProcessGroup;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
public interface ControllerServiceNode extends ComponentNode,
VersionedComponent {
/**
@@ -182,6 +183,15 @@ public interface ControllerServiceNode extends
ComponentNode, VersionedComponent
boolean isActive();
/**
+ * Waits up to the given amount of time for the Controller Service to
transition to an ENABLED state.
+ * @param timePeriod maximum amount of time to wait
+ * @param timeUnit the unit for the time period
+ * @return <code>true</code> if the Controller Service finished enabling,
<code>false</code> otherwise
+ * @throws InterruptedException if interrupted while waiting for the
service complete its enabling
+ */
+ boolean awaitEnabled(long timePeriod, TimeUnit timeUnit) throws
InterruptedException;
+
+ /**
* Sets a new proxy and implementation for this node.
*
* @param implementation the actual implementation controller service
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
index 5fb2331..56cffe2 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
@@ -16,18 +16,14 @@
*/
package org.apache.nifi.controller;
-import java.lang.reflect.Proxy;
-import java.net.URL;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.RequiresInstanceClassLoading;
import org.apache.nifi.annotation.configuration.DefaultSettings;
import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.ConfigurableComponent;
+import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.validation.ValidationTrigger;
@@ -45,6 +41,7 @@ import
org.apache.nifi.controller.service.StandardControllerServiceInvocationHan
import org.apache.nifi.controller.service.StandardControllerServiceNode;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.GhostProcessor;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.ProcessorInitializationContext;
@@ -62,6 +59,14 @@ import org.apache.nifi.scheduling.SchedulingStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.Proxy;
+import java.net.URL;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
public class ExtensionBuilder {
private static final Logger logger =
LoggerFactory.getLogger(ExtensionBuilder.class);
@@ -362,6 +367,7 @@ public class ExtensionBuilder {
final Class<? extends ControllerService> controllerServiceClass =
rawClass.asSubclass(ControllerService.class);
final ControllerService serviceImpl =
controllerServiceClass.newInstance();
+
final StandardControllerServiceInvocationHandler invocationHandler
= new StandardControllerServiceInvocationHandler(extensionManager, serviceImpl);
// extract all interfaces... controllerServiceClass is non null so
getAllInterfaces is non null
@@ -384,6 +390,8 @@ public class ExtensionBuilder {
serviceProvider, stateManager, kerberosConfig,
nodeTypeProvider);
serviceImpl.initialize(initContext);
+ verifyControllerServiceReferences(serviceImpl,
bundle.getClassLoader());
+
final LoggableComponent<ControllerService>
originalLoggableComponent = new LoggableComponent<>(serviceImpl,
bundleCoordinate, terminationAwareLogger);
final LoggableComponent<ControllerService>
proxiedLoggableComponent = new LoggableComponent<>(proxiedService,
bundleCoordinate, terminationAwareLogger);
@@ -402,6 +410,68 @@ public class ExtensionBuilder {
}
}
+ private static void verifyControllerServiceReferences(final
ConfigurableComponent component, final ClassLoader bundleClassLoader) throws
InstantiationException {
+ // If a component lives in the same NAR as a Controller Service API,
and the component references the Controller Service API (either
+ // by itself implementing the API or by having a Property Descriptor
that identifies the Controller Service), then the component is not
+ // allowed to Require Instance Class Loading. This is done because
when a component requires Instance Class Loading, the jars within the
+ // NAR and its parents must be copied to a new class loader all the
way up to the point of the Controller Service APIs. If the Controller
+ // Service API lives in the same NAR as the implementation itself,
then we cannot duplicate the NAR ClassLoader. Otherwise, we would have
+ // two different NAR ClassLoaders that each define the Service API.
And the Service API class must live in the parent ClassLoader for both
+ // the referencing component AND the implementing component.
+
+ // if the extension does not require instance classloading, there is
no concern.
+ final boolean requiresInstanceClassLoading =
component.getClass().isAnnotationPresent(RequiresInstanceClassLoading.class);
+ if (!requiresInstanceClassLoading) {
+ logger.debug("Instance ClassLoading is not required for {}",
component);
+ return;
+ }
+
+ logger.debug("Component {} requires Instance Class Loading",
component);
+
+ final Class<?> originalExtensionType = component.getClass();
+ final ClassLoader originalExtensionClassLoader =
originalExtensionType.getClassLoader();
+
+ // Find any Controller Service API's that are bundled in the same NAR.
+ final Set<Class<?>> cobundledApis = new HashSet<>();
+ try (final NarCloseable closeable =
NarCloseable.withComponentNarLoader(component.getClass().getClassLoader())) {
+ final List<PropertyDescriptor> descriptors =
component.getPropertyDescriptors();
+ if (descriptors != null && !descriptors.isEmpty()) {
+ for (final PropertyDescriptor descriptor : descriptors) {
+ final Class<? extends ControllerService> serviceApi =
descriptor.getControllerServiceDefinition();
+ if (serviceApi != null &&
bundleClassLoader.equals(serviceApi.getClassLoader())) {
+ cobundledApis.add(serviceApi);
+ }
+ }
+ }
+ }
+
+ logger.debug("Component {} is co-bundled with {} Controller Service
APIs based on referenced Controller Services: {}", component,
cobundledApis.size(), cobundledApis);
+
+ // If the component is a Controller Service, it should also not extend
from any API that is in the same class loader.
+ if (component instanceof ControllerService) {
+ Class<?> extensionType = component.getClass();
+ while (extensionType != null) {
+ for (final Class<?> ifc : extensionType.getInterfaces()) {
+ if
(originalExtensionClassLoader.equals(ifc.getClassLoader())) {
+ cobundledApis.add(ifc);
+ }
+ }
+
+ extensionType = extensionType.getSuperclass();
+ }
+
+ logger.debug("Component {} is co-bundled with {} Controller
Service APIs based on referenced Controller Services and services that are
implemented: {}",
+ component, cobundledApis.size(), cobundledApis);
+ }
+
+ if (!cobundledApis.isEmpty()) {
+ final String message = String.format("Controller Service %s is
bundled with its supporting APIs %s. The service APIs should not be bundled
with the implementations.",
+ originalExtensionType.getName(),
org.apache.nifi.util.StringUtils.join(cobundledApis.stream().map(Class::getName).collect(Collectors.toSet()),
", "));
+ throw new InstantiationException(message);
+ }
+ }
+
+
private ControllerServiceNode createGhostControllerServiceNode() {
final String simpleClassName = type.contains(".") ?
StringUtils.substringAfterLast(type, ".") : type;
final String componentType = "(Missing) " + simpleClassName;
@@ -422,10 +492,14 @@ public class ExtensionBuilder {
private LoggableComponent<Processor> createLoggableProcessor() throws
ProcessorInstantiationException {
try {
final LoggableComponent<Processor> processorComponent =
createLoggableComponent(Processor.class);
+ final Processor processor = processorComponent.getComponent();
final ProcessorInitializationContext initiContext = new
StandardProcessorInitializationContext(identifier,
processorComponent.getLogger(),
serviceProvider, nodeTypeProvider, kerberosConfig);
- processorComponent.getComponent().initialize(initiContext);
+ processor.initialize(initiContext);
+
+ final Bundle bundle = extensionManager.getBundle(bundleCoordinate);
+ verifyControllerServiceReferences(processor,
bundle.getClassLoader());
return processorComponent;
} catch (final Exception e) {
@@ -444,6 +518,9 @@ public class ExtensionBuilder {
taskComponent.getComponent().initialize(config);
+ final Bundle bundle = extensionManager.getBundle(bundleCoordinate);
+ verifyControllerServiceReferences(taskComponent.getComponent(),
bundle.getClassLoader());
+
return taskComponent;
} catch (final Exception e) {
throw new ReportingTaskInstantiationException(type, e);
@@ -463,6 +540,7 @@ public class ExtensionBuilder {
Thread.currentThread().setContextClassLoader(detectedClassLoader);
final Object extensionInstance = rawClass.newInstance();
+
final ComponentLog componentLog = new
SimpleProcessLogger(identifier, extensionInstance);
final TerminationAwareLogger terminationAwareLogger = new
TerminationAwareLogger(componentLog);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index d32d431..16d3347 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -141,6 +141,7 @@ import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.groups.StandardProcessGroup;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.nar.NarThreadContextClassLoader;
@@ -576,7 +577,7 @@ public class FlowController implements
ReportingTaskProvider, Authorizable, Node
this.reloadComponent = new StandardReloadComponent(this);
final ProcessGroup rootGroup = new
StandardProcessGroup(ComponentIdGenerator.generateId().toString(),
controllerServiceProvider, processScheduler,
- encryptor, extensionManager, stateManagerProvider,
flowManager, flowRegistryClient, reloadComponent, new
MutableVariableRegistry(this.variableRegistry), this);
+ encryptor, extensionManager, stateManagerProvider, flowManager,
flowRegistryClient, reloadComponent, new
MutableVariableRegistry(this.variableRegistry), this);
rootGroup.setName(FlowManager.DEFAULT_ROOT_GROUP_NAME);
setRootGroup(rootGroup);
instanceId = ComponentIdGenerator.generateId().toString();
@@ -1626,22 +1627,23 @@ public class FlowController implements
ReportingTaskProvider, Authorizable, Node
public void verifyComponentTypesInSnippet(final VersionedProcessGroup
versionedFlow) {
final Map<String, Set<BundleCoordinate>> processorClasses = new
HashMap<>();
- for (final Class<?> c :
extensionManager.getExtensions(Processor.class)) {
- final String name = c.getName();
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(Processor.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
processorClasses.put(name,
extensionManager.getBundles(name).stream().map(bundle ->
bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
}
verifyProcessorsInVersionedFlow(versionedFlow, processorClasses);
final Map<String, Set<BundleCoordinate>> controllerServiceClasses =
new HashMap<>();
- for (final Class<?> c :
extensionManager.getExtensions(ControllerService.class)) {
- final String name = c.getName();
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(ControllerService.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
controllerServiceClasses.put(name,
extensionManager.getBundles(name).stream().map(bundle ->
bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
}
verifyControllerServicesInVersionedFlow(versionedFlow,
controllerServiceClasses);
final Set<String> prioritizerClasses = new HashSet<>();
- for (final Class<?> c :
extensionManager.getExtensions(FlowFilePrioritizer.class)) {
- prioritizerClasses.add(c.getName());
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(FlowFilePrioritizer.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
+ prioritizerClasses.add(name);
}
final Set<VersionedConnection> allConns = new HashSet<>();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSnippet.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSnippet.java
index 215ec89..058d051 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSnippet.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSnippet.java
@@ -37,6 +37,7 @@ import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroupPortDescriptor;
import org.apache.nifi.logging.LogLevel;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.parameter.ParameterContext;
import org.apache.nifi.processor.Processor;
@@ -113,22 +114,23 @@ public class StandardFlowSnippet implements FlowSnippet {
public void verifyComponentTypesInSnippet() {
final Map<String, Set<BundleCoordinate>> processorClasses = new
HashMap<>();
- for (final Class<?> c :
extensionManager.getExtensions(Processor.class)) {
- final String name = c.getName();
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(Processor.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
processorClasses.put(name,
extensionManager.getBundles(name).stream().map(bundle ->
bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
}
verifyProcessorsInSnippet(dto, processorClasses);
final Map<String, Set<BundleCoordinate>> controllerServiceClasses =
new HashMap<>();
- for (final Class<?> c :
extensionManager.getExtensions(ControllerService.class)) {
- final String name = c.getName();
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(ControllerService.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
controllerServiceClasses.put(name,
extensionManager.getBundles(name).stream().map(bundle ->
bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
}
verifyControllerServicesInSnippet(dto, controllerServiceClasses);
final Set<String> prioritizerClasses = new HashSet<>();
- for (final Class<?> c :
extensionManager.getExtensions(FlowFilePrioritizer.class)) {
- prioritizerClasses.add(c.getName());
+ for (final ExtensionDefinition extensionDefinition :
extensionManager.getExtensions(FlowFilePrioritizer.class)) {
+ final String name =
extensionDefinition.getImplementationClassName();
+ prioritizerClasses.add(name);
}
final Set<ConnectionDTO> allConns = new HashSet<>();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/provenance/ComponentIdentifierLookup.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/provenance/ComponentIdentifierLookup.java
index f5da0fc..6017eff 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/provenance/ComponentIdentifierLookup.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/provenance/ComponentIdentifierLookup.java
@@ -16,9 +16,11 @@
*/
package org.apache.nifi.provenance;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.processor.Processor;
import java.util.ArrayList;
@@ -46,14 +48,15 @@ public class ComponentIdentifierLookup implements
IdentifierLookup {
@Override
public List<String> getComponentTypes() {
- final Set<Class> procClasses =
flowController.getExtensionManager().getExtensions(Processor.class);
+ final Set<ExtensionDefinition> procDefinitions =
flowController.getExtensionManager().getExtensions(Processor.class);
- final List<String> componentTypes = new ArrayList<>(procClasses.size()
+ 2);
+ final List<String> componentTypes = new
ArrayList<>(procDefinitions.size() + 2);
componentTypes.add(ProvenanceEventRecord.REMOTE_INPUT_PORT_TYPE);
componentTypes.add(ProvenanceEventRecord.REMOTE_OUTPUT_PORT_TYPE);
- procClasses.stream()
- .map(Class::getSimpleName)
+ procDefinitions.stream()
+ .map(ExtensionDefinition::getImplementationClassName)
+ .map(className -> className.contains(".") ?
StringUtils.substringAfterLast(className, ".") : className)
.forEach(componentTypes::add);
return componentTypes;
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/DirectInjectionExtensionManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/DirectInjectionExtensionManager.java
index 56cc212..e0343f6 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/DirectInjectionExtensionManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/DirectInjectionExtensionManager.java
@@ -44,7 +44,7 @@ public class DirectInjectionExtensionManager extends
StandardExtensionDiscoverin
throw new IllegalArgumentException("Given extension is not a
Processor, Controller Service, or Reporting Task");
}
- super.loadExtension(extension, extensionType, INTEGRATION_TEST_BUNDLE);
+ super.loadExtension(extension.getClass().getName(), extensionType,
INTEGRATION_TEST_BUNDLE);
}
public void injectExtensionType(final Class<?> extensionType, final String
implementationClassName) {
@@ -59,6 +59,6 @@ public class DirectInjectionExtensionManager extends
StandardExtensionDiscoverin
}
public void injectExtensionType(final Class<?> extensionType, final
Class<?> implementationClass) {
- super.registerExtensionClass(extensionType, implementationClass,
INTEGRATION_TEST_BUNDLE);
+ super.registerExtensionClass(extensionType,
implementationClass.getName(), INTEGRATION_TEST_BUNDLE);
}
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-loading-utils/src/main/java/org/apache/nifi/nar/StandardNarLoader.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-loading-utils/src/main/java/org/apache/nifi/nar/StandardNarLoader.java
index 99ef8f3..b027dca 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-loading-utils/src/main/java/org/apache/nifi/nar/StandardNarLoader.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-loading-utils/src/main/java/org/apache/nifi/nar/StandardNarLoader.java
@@ -112,14 +112,14 @@ public class StandardNarLoader implements NarLoader {
// Call the DocGenerator for the classes that were loaded from
each Bundle
for (final Bundle bundle : loadedBundles) {
final BundleCoordinate bundleCoordinate =
bundle.getBundleDetails().getCoordinate();
- final Set<Class> extensions =
extensionManager.getTypes(bundleCoordinate);
- if (extensions.isEmpty()) {
+ final Set<ExtensionDefinition> extensionDefinitions =
extensionManager.getTypes(bundleCoordinate);
+ if (extensionDefinitions.isEmpty()) {
LOGGER.debug("No documentation to generate for {} because
no extensions were found",
new Object[]{bundleCoordinate.getCoordinate()});
} else {
LOGGER.debug("Generating documentation for {} extensions
in {}",
- new Object[]{extensions.size(),
bundleCoordinate.getCoordinate()});
- DocGenerator.documentConfigurableComponent(extensions,
docsWorkingDir, extensionManager);
+ new Object[]{extensionDefinitions.size(),
bundleCoordinate.getCoordinate()});
+
DocGenerator.documentConfigurableComponent(extensionDefinitions,
docsWorkingDir, extensionManager);
}
}
@@ -161,7 +161,7 @@ public class StandardNarLoader implements NarLoader {
return null;
}
- final File unpackedExtension = NarUnpacker.unpackNar(narFile,
extensionsWorkingDir);
+ final File unpackedExtension = NarUnpacker.unpackNar(narFile,
extensionsWorkingDir, true);
NarUnpacker.mapExtension(unpackedExtension, coordinate,
docsWorkingDir, extensionMapping);
return unpackedExtension;
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionDefinition.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionDefinition.java
new file mode 100644
index 0000000..a8f3ce3
--- /dev/null
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionDefinition.java
@@ -0,0 +1,88 @@
+/*
+ * 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.nar;
+
+import org.apache.nifi.bundle.Bundle;
+
+import java.util.Objects;
+
+/**
+ * Provides a wrapper for the elements that make an extension unique. For
example, there may be a class named org.apache.nifi.extensions.ExtensionABC
+ * that exists in a given NiFi Archive (NAR). There may also be several
classes named org.apache.nifi.extensions.ExtensionABC, each in its own NAR,
+ * and each of those would be a separate extension. This class provides a
mechanism by which the relevant bits to determine an extension's uniqueness
+ * can be bundled together into a single class.
+ */
+public class ExtensionDefinition {
+ private final String implementationClassName;
+ private final Bundle bundle;
+ private final Class<?> extensionType;
+
+ public ExtensionDefinition(final String implementationClassName, final
Bundle bundle, final Class<?> extensionType) {
+ this.implementationClassName = implementationClassName;
+ this.bundle = bundle;
+ this.extensionType = extensionType;
+ }
+
+ /**
+ * @return the fully qualified class name of the class that implements the
extension
+ */
+ public String getImplementationClassName() {
+ return implementationClassName;
+ }
+
+ /**
+ * @return the Bundle that contains the extension
+ */
+ public Bundle getBundle() {
+ return bundle;
+ }
+
+ /**
+ * @return the type of Extension (e.g., {@link
org.apache.nifi.processor.Processor}, {@link
org.apache.nifi.controller.ControllerService},
+ * or {@link org.apache.nifi.reporting.ReportingTask}.
+ */
+ public Class<?> getExtensionType() {
+ return extensionType;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ final ExtensionDefinition that = (ExtensionDefinition) o;
+ return Objects.equals(implementationClassName,
that.implementationClassName)
+ && Objects.equals(bundle, that.bundle)
+ && Objects.equals(extensionType, that.extensionType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(implementationClassName, bundle, extensionType);
+ }
+
+ @Override
+ public String toString() {
+ return "ExtensionDefinition[type=" + extensionType.getSimpleName() +
", implementation=" + implementationClassName + ", bundle=" + bundle + "]";
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionManager.java
index 5b9f2e6..61c96c7 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/ExtensionManager.java
@@ -90,9 +90,16 @@ public interface ExtensionManager {
* Retrieves the extension classes that were loaded from the bundle with
the given coordinate.
*
* @param bundleCoordinate the coordinate
- * @return the classes from the bundle with that coordinate
+ * @return the definitions of the extensions from the bundle with that
coordinate
*/
- Set<Class> getTypes(BundleCoordinate bundleCoordinate);
+ Set<ExtensionDefinition> getTypes(BundleCoordinate bundleCoordinate);
+
+ /**
+ * Returns the Class that is described by the given definition
+ * @param extensionDefinition the extension definition
+ * @return the extension's class
+ */
+ Class<?> getClass(ExtensionDefinition extensionDefinition);
/**
* Retrieves the bundle for the given class loader.
@@ -108,9 +115,9 @@ public interface ExtensionManager {
* (i.e getExtensions(Processor.class)
*
* @param definition the extension definition, such as Processor.class
- * @return the set of extensions implementing the defintion
+ * @return the set of extension definitions that describe the the
extensions implementing the defintion
*/
- Set<Class> getExtensions(Class<?> definition);
+ Set<ExtensionDefinition> getExtensions(Class<?> definition);
/**
* Gets the temp component with the given type from the given bundle.
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java
index a6c5a8f..959a38e 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java
@@ -43,19 +43,23 @@ import org.apache.nifi.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@@ -71,15 +75,14 @@ public class StandardExtensionDiscoveringManager implements
ExtensionDiscovering
private static final Logger logger =
LoggerFactory.getLogger(StandardExtensionDiscoveringManager.class);
// Maps a service definition (interface) to those classes that implement
the interface
- private final Map<Class, Set<Class>> definitionMap = new HashMap<>();
+ private final Map<Class, Set<ExtensionDefinition>> definitionMap = new
HashMap<>();
private final Map<String, List<Bundle>> classNameBundleLookup = new
HashMap<>();
- private final Map<BundleCoordinate, Set<Class>>
bundleCoordinateClassesLookup = new HashMap<>();
+ private final Map<BundleCoordinate, Set<ExtensionDefinition>>
bundleCoordinateClassesLookup = new HashMap<>();
private final Map<BundleCoordinate, Bundle> bundleCoordinateBundleLookup =
new HashMap<>();
private final Map<ClassLoader, Bundle> classLoaderBundleLookup = new
HashMap<>();
private final Map<String, ConfigurableComponent> tempComponentLookup = new
HashMap<>();
- private final Map<String, Class<?>> requiresInstanceClassLoading = new
HashMap<>();
private final Map<String, InstanceClassLoader> instanceClassloaderLookup =
new ConcurrentHashMap<>();
public StandardExtensionDiscoveringManager() {
@@ -128,7 +131,11 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
// so that static initialization techniques that depend on the
context class loader will work properly
final ClassLoader ncl = bundle.getClassLoader();
Thread.currentThread().setContextClassLoader(ncl);
+
+ final long loadStart = System.currentTimeMillis();
loadExtensions(bundle);
+ final long loadMillis = System.currentTimeMillis() - loadStart;
+ logger.info("Loaded extensions for {} in {} millis",
bundle.getBundleDetails(), loadMillis);
// Create a look-up from coordinate to bundle
bundleCoordinateBundleLookup.put(bundle.getBundleDetails().getCoordinate(),
bundle);
@@ -145,74 +152,103 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
*
* @param bundle from which to load extensions
*/
- @SuppressWarnings("unchecked")
private void loadExtensions(final Bundle bundle) {
- for (final Map.Entry<Class, Set<Class>> entry :
definitionMap.entrySet()) {
- final boolean isControllerService =
ControllerService.class.equals(entry.getKey());
- final boolean isProcessor = Processor.class.equals(entry.getKey());
- final boolean isReportingTask =
ReportingTask.class.equals(entry.getKey());
-
- final ServiceLoader<?> serviceLoader =
ServiceLoader.load(entry.getKey(), bundle.getClassLoader());
- for (final Object o : serviceLoader) {
- try {
- loadExtension(o, entry.getKey(), bundle);
- } catch (Exception e) {
- logger.warn("Failed to register extension {} due to: {}" ,
new Object[]{o.getClass().getCanonicalName(), e.getMessage()});
- if (logger.isDebugEnabled()) {
- logger.debug("", e);
+ for (final Class extensionType : definitionMap.keySet()) {
+ final String serviceType = extensionType.getName();
+
+ try {
+ final Set<URL> serviceResourceUrls =
getServiceFileURLs(bundle, extensionType);
+ logger.debug("Bundle {} has the following Services File URLs
for {}: {}", bundle, serviceType, serviceResourceUrls);
+
+ for (final URL serviceResourceUrl : serviceResourceUrls) {
+ final Set<String> implementationClassNames =
getServiceFileImplementationClassNames(serviceResourceUrl);
+ logger.debug("Bundle {} defines {} implementations of
interface {}", bundle, implementationClassNames.size(), serviceType);
+
+ for (final String implementationClassName :
implementationClassNames) {
+ try {
+ loadExtension(implementationClassName,
extensionType, bundle);
+ logger.debug("Successfully loaded {} {} from {}",
extensionType.getSimpleName(), implementationClassName, bundle);
+ } catch (final Exception e) {
+ logger.error("Failed to register {} of type {} in
bundle {}" , extensionType.getSimpleName(), implementationClassName, bundle, e);
+ }
}
}
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to get resources of type "
+ serviceType + " from bundle " + bundle);
}
-
- classLoaderBundleLookup.put(bundle.getClassLoader(), bundle);
}
+
+ classLoaderBundleLookup.put(bundle.getClassLoader(), bundle);
}
- protected void loadExtension(final Object extension, final Class<?>
extensionType, final Bundle bundle) {
- final boolean isControllerService =
ControllerService.class.equals(extensionType);
- final boolean isProcessor = Processor.class.equals(extensionType);
- final boolean isReportingTask =
ReportingTask.class.equals(extensionType);
+ private Set<String> getServiceFileImplementationClassNames(final URL
serviceFileUrl) throws IOException {
+ final Set<String> implementationClassNames = new HashSet<>();
- // create a cache of temp ConfigurableComponent instances, the
initialize here has to happen before the checks below
- if ((isControllerService || isProcessor || isReportingTask) &&
extension instanceof ConfigurableComponent) {
- final ConfigurableComponent configurableComponent =
(ConfigurableComponent) extension;
- initializeTempComponent(configurableComponent);
+ try (final InputStream in = serviceFileUrl.openStream();
+ final Reader inputStreamReader = new InputStreamReader(in);
+ final BufferedReader reader = new
BufferedReader(inputStreamReader)) {
- final String cacheKey =
getClassBundleKey(extension.getClass().getCanonicalName(),
bundle.getBundleDetails().getCoordinate());
- tempComponentLookup.put(cacheKey, configurableComponent);
- }
+ String line;
+ while ((line = reader.readLine()) != null) {
+ // Remove anything after the #
+ final int index = line.indexOf("#");
+ if (index >= 0) {
+ line = line.substring(0, index);
+ }
- // only consider extensions discovered directly in this bundle
- boolean registerExtension =
bundle.getClassLoader().equals(extension.getClass().getClassLoader());
+ // Ignore empty line
+ line = line.trim();
+ if (line.isEmpty()) {
+ continue;
+ }
- if (registerExtension) {
- final Class<?> extensionClass = extension.getClass();
- if (isControllerService &&
!checkControllerServiceEligibility(extensionClass)) {
- registerExtension = false;
- logger.error(String.format(
- "Skipping Controller Service %s because it is bundled with
its supporting APIs and requires instance class loading.",
extensionClass.getName()));
+ implementationClassNames.add(line);
}
+ }
- final boolean canReferenceControllerService = (isControllerService
|| isProcessor || isReportingTask) && extension instanceof
ConfigurableComponent;
- if (canReferenceControllerService &&
!checkControllerServiceReferenceEligibility((ConfigurableComponent) extension,
bundle.getClassLoader())) {
- registerExtension = false;
- logger.error(String.format(
- "Skipping component %s because it is bundled with its
referenced Controller Service APIs and requires instance class loading.",
extensionClass.getName()));
- }
+ return implementationClassNames;
+ }
- if (registerExtension) {
- registerExtensionClass(extensionType, extension.getClass(),
bundle);
- }
+ /**
+ * Returns a Set of URL's for all Service Files (i.e.,
META-INF/services/<interface name> files)
+ * that define the extensions that exist for the given bundle. The
returned set will only contain URL's for
+ * which the services file live in the given bundle directly and NOT the
parent/ancestor bundle.
+ *
+ * @param bundle the bundle whose extensions are of interest
+ * @param extensionType the type of extension (I.e., Processor,
ControllerService, ReportingTask, etc.)
+ * @return the set of URL's that point to Service Files for the given
extension type in the given bundle. An empty set will be
+ * returned if no service files exist
+ *
+ * @throws IOException if unable to read the services file from the given
bundle's classloader.
+ */
+ private Set<URL> getServiceFileURLs(final Bundle bundle, final Class<?>
extensionType) throws IOException {
+ final String servicesFile = "META-INF/services/" +
extensionType.getName();
+
+ final Enumeration<URL> serviceResourceUrlEnum =
bundle.getClassLoader().getResources(servicesFile);
+ final Set<URL> serviceResourceUrls = new HashSet<>();
+ while (serviceResourceUrlEnum.hasMoreElements()) {
+ serviceResourceUrls.add(serviceResourceUrlEnum.nextElement());
+ }
+
+ final Enumeration<URL> parentResourceUrlEnum =
bundle.getClassLoader().getParent().getResources(servicesFile);
+ while (parentResourceUrlEnum.hasMoreElements()) {
+ serviceResourceUrls.remove(parentResourceUrlEnum.nextElement());
}
+
+ return serviceResourceUrls;
}
- protected void registerExtensionClass(final Class<?> extensionType, final
Class<?> implementationClass, final Bundle bundle) {
- final Set<Class> registeredClasses = definitionMap.get(extensionType);
- registerServiceClass(implementationClass, classNameBundleLookup,
bundleCoordinateClassesLookup, bundle, registeredClasses);
+ protected void loadExtension(final String extensionClassName, final
Class<?> extensionType, final Bundle bundle) {
+ registerExtensionClass(extensionType, extensionClassName, bundle);
}
+ protected void registerExtensionClass(final Class<?> extensionType, final
String implementationClassName, final Bundle bundle) {
+ final Set<ExtensionDefinition> registeredClasses =
definitionMap.get(extensionType);
+ registerServiceClass(implementationClassName, extensionType,
classNameBundleLookup, bundleCoordinateClassesLookup, bundle,
registeredClasses);
+ }
- private void initializeTempComponent(final ConfigurableComponent
configurableComponent) {
+
+ protected void initializeTempComponent(final ConfigurableComponent
configurableComponent) {
ConfigurableComponentInitializer initializer = null;
try {
initializer =
ConfigurableComponentInitializerFactory.createComponentInitializer(this,
configurableComponent.getClass());
@@ -222,78 +258,28 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
}
}
- private static boolean checkControllerServiceReferenceEligibility(final
ConfigurableComponent component, final ClassLoader classLoader) {
- // if the extension does not require instance classloading, its
eligible
- final boolean requiresInstanceClassLoading =
component.getClass().isAnnotationPresent(RequiresInstanceClassLoading.class);
-
- final Set<Class> cobundledApis = new HashSet<>();
- try (final NarCloseable closeable =
NarCloseable.withComponentNarLoader(component.getClass().getClassLoader())) {
- final List<PropertyDescriptor> descriptors =
component.getPropertyDescriptors();
- if (descriptors != null && !descriptors.isEmpty()) {
- for (final PropertyDescriptor descriptor : descriptors) {
- final Class<? extends ControllerService> serviceApi =
descriptor.getControllerServiceDefinition();
- if (serviceApi != null &&
classLoader.equals(serviceApi.getClassLoader())) {
- cobundledApis.add(serviceApi);
- }
- }
- }
- }
-
- if (!cobundledApis.isEmpty()) {
- logger.warn(String.format(
- "Component %s is bundled with its referenced Controller
Service APIs %s. The service APIs should not be bundled with component
implementations that reference it.",
- component.getClass().getName(),
StringUtils.join(cobundledApis.stream().map(Class::getName).collect(Collectors.toSet()),
", ")));
- }
-
- // the component is eligible when it does not require instance
classloading or when the supporting APIs are bundled in a parent NAR
- return requiresInstanceClassLoading == false ||
cobundledApis.isEmpty();
- }
-
- private static boolean checkControllerServiceEligibility(Class
extensionType) {
- final Class originalExtensionType = extensionType;
- final ClassLoader originalExtensionClassLoader =
extensionType.getClassLoader();
-
- // if the extension does not require instance classloading, its
eligible
- final boolean requiresInstanceClassLoading =
extensionType.isAnnotationPresent(RequiresInstanceClassLoading.class);
-
- final Set<Class> cobundledApis = new HashSet<>();
- while (extensionType != null) {
- for (final Class i : extensionType.getInterfaces()) {
- if (originalExtensionClassLoader.equals(i.getClassLoader())) {
- cobundledApis.add(i);
- }
- }
-
- extensionType = extensionType.getSuperclass();
- }
-
- if (!cobundledApis.isEmpty()) {
- logger.warn(String.format("Controller Service %s is bundled with
its supporting APIs %s. The service APIs should not be bundled with the
implementations.",
- originalExtensionType.getName(),
StringUtils.join(cobundledApis.stream().map(Class::getName).collect(Collectors.toSet()),
", ")));
- }
-
- // the service is eligible when it does not require instance
classloading or when the supporting APIs are bundled in a parent NAR
- return requiresInstanceClassLoading == false ||
cobundledApis.isEmpty();
+ protected void addTempComponent(final ConfigurableComponent instance,
final BundleCoordinate coordinate) {
+ final String cacheKey =
getClassBundleKey(instance.getClass().getCanonicalName(), coordinate);
+ tempComponentLookup.put(cacheKey, instance);
}
/**
* Registers extension for the specified type from the specified Bundle.
*
- * @param type the extension type
+ * @param className the fully qualified class name of the extension
implementation
* @param classNameBundleMap mapping of classname to Bundle
* @param bundle the Bundle being mapped to
* @param classes to map to this classloader but which come from its
ancestors
*/
- private void registerServiceClass(final Class<?> type,
+ private void registerServiceClass(final String className, final Class<?>
extensionType,
final Map<String, List<Bundle>>
classNameBundleMap,
- final Map<BundleCoordinate,
Set<Class>> bundleCoordinateClassesMap,
- final Bundle bundle, final
Set<Class> classes) {
- final String className = type.getName();
+ final Map<BundleCoordinate,
Set<ExtensionDefinition>> bundleCoordinateClassesMap,
+ final Bundle bundle, final
Set<ExtensionDefinition> classes) {
final BundleCoordinate bundleCoordinate =
bundle.getBundleDetails().getCoordinate();
// get the bundles that have already been registered for the class name
final List<Bundle> registeredBundles =
classNameBundleMap.computeIfAbsent(className, (key) -> new ArrayList<>());
- final Set<Class> bundleCoordinateClasses =
bundleCoordinateClassesMap.computeIfAbsent(bundleCoordinate, (key) -> new
HashSet<>());
+ final Set<ExtensionDefinition> bundleExtensionDefinitions =
bundleCoordinateClassesMap.computeIfAbsent(bundleCoordinate, (key) -> new
HashSet<>());
boolean alreadyRegistered = false;
for (final Bundle registeredBundle : registeredBundles) {
@@ -307,27 +293,33 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
// if the type wasn't loaded from an ancestor, and the type isn't
a processor, cs, or reporting task, then
// fail registration because we don't support multiple versions of
any other types
- if (!multipleVersionsAllowed(type)) {
- throw new IllegalStateException("Attempt was made to load " +
className + " from "
- +
bundle.getBundleDetails().getCoordinate().getCoordinate()
- + " but that class name is already loaded/registered
from " + registeredBundle.getBundleDetails().getCoordinate()
- + " and multiple versions are not supported for this
type"
- );
+ if (!multipleVersionsAllowed(extensionType)) {
+ logger.debug("Attempt was made to load {} from {} but that
class name is already loaded/registered from {} and multiple versions are not
supported for this type",
+ className,
bundle.getBundleDetails().getCoordinate().getCoordinate(),
registeredBundle.getBundleDetails().getCoordinate());
+ return;
}
}
// if none of the above was true then register the new bundle
if (!alreadyRegistered) {
registeredBundles.add(bundle);
- bundleCoordinateClasses.add(type);
- classes.add(type);
- if (type.isAnnotationPresent(RequiresInstanceClassLoading.class)) {
- final String cacheKey = getClassBundleKey(className,
bundleCoordinate);
- requiresInstanceClassLoading.put(cacheKey, type);
- }
+ final ExtensionDefinition extensionDefinition = new
ExtensionDefinition(className, bundle, extensionType);
+ bundleExtensionDefinitions.add(extensionDefinition);
+ classes.add(extensionDefinition);
}
+ }
+ @Override
+ public Class<?> getClass(final ExtensionDefinition extensionDefinition) {
+ final Bundle bundle = extensionDefinition.getBundle();
+ final ClassLoader bundleClassLoader = bundle.getClassLoader();
+
+ try (final NarCloseable x =
NarCloseable.withComponentNarLoader(bundleClassLoader)) {
+ return
Class.forName(extensionDefinition.getImplementationClassName(), true,
bundleClassLoader);
+ } catch (final Exception e) {
+ throw new RuntimeException("Could not create Class for " +
extensionDefinition, e);
+ }
}
/**
@@ -338,6 +330,22 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
return Processor.class.isAssignableFrom(type) ||
ControllerService.class.isAssignableFrom(type) ||
ReportingTask.class.isAssignableFrom(type);
}
+ protected boolean isInstanceClassLoaderRequired(final String classType,
final Bundle bundle) {
+ // We require instance Class Loaders if the component has the
@RequiresInstanceClassLoader annotation and is loaded from the NAR ClassLoader.
+ // So the first check is to see if the bundle's ClassLoader is a
NarClassLoader.
+ final ClassLoader bundleClassLoader = bundle.getClassLoader();
+ if (!(bundleClassLoader instanceof NarClassLoader)) {
+ return false;
+ }
+
+ final ConfigurableComponent tempComponent =
getTempComponent(classType, bundle.getBundleDetails().getCoordinate());
+ if (tempComponent == null) {
+ return false;
+ }
+
+ return
tempComponent.getClass().isAnnotationPresent(RequiresInstanceClassLoading.class);
+ }
+
@Override
public InstanceClassLoader createInstanceClassLoader(final String
classType, final String instanceIdentifier, final Bundle bundle, final Set<URL>
additionalUrls) {
if (StringUtils.isEmpty(classType)) {
@@ -358,10 +366,12 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
InstanceClassLoader instanceClassLoader;
final ClassLoader bundleClassLoader = bundle.getClassLoader();
- final String key = getClassBundleKey(classType,
bundle.getBundleDetails().getCoordinate());
- if (requiresInstanceClassLoading.containsKey(key) && bundleClassLoader
instanceof NarClassLoader) {
- final Class<?> type = requiresInstanceClassLoading.get(key);
+ final boolean requiresInstanceClassLoader =
isInstanceClassLoaderRequired(classType, bundle);
+ if (requiresInstanceClassLoader) {
+ final ConfigurableComponent tempComponent =
getTempComponent(classType, bundle.getBundleDetails().getCoordinate());
+ final Class<?> type = tempComponent.getClass();
+
final RequiresInstanceClassLoading requiresInstanceClassLoading =
type.getAnnotation(RequiresInstanceClassLoading.class);
final NarClassLoader narBundleClassLoader = (NarClassLoader)
bundleClassLoader;
@@ -469,6 +479,7 @@ public class StandardExtensionDiscoveringManager implements
ExtensionDiscovering
if (classType == null) {
throw new IllegalArgumentException("Class type cannot be null");
}
+
final List<Bundle> bundles = classNameBundleLookup.get(classType);
return bundles == null ? Collections.emptyList() : new
ArrayList<>(bundles);
}
@@ -482,11 +493,11 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
}
@Override
- public Set<Class> getTypes(final BundleCoordinate bundleCoordinate) {
+ public Set<ExtensionDefinition> getTypes(final BundleCoordinate
bundleCoordinate) {
if (bundleCoordinate == null) {
throw new IllegalArgumentException("BundleCoordinate cannot be
null");
}
- final Set<Class> types =
bundleCoordinateClassesLookup.get(bundleCoordinate);
+ final Set<ExtensionDefinition> types =
bundleCoordinateClassesLookup.get(bundleCoordinate);
return types == null ? Collections.emptySet() :
Collections.unmodifiableSet(types);
}
@@ -499,16 +510,16 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
}
@Override
- public Set<Class> getExtensions(final Class<?> definition) {
+ public Set<ExtensionDefinition> getExtensions(final Class<?> definition) {
if (definition == null) {
throw new IllegalArgumentException("Class cannot be null");
}
- final Set<Class> extensions = definitionMap.get(definition);
+ final Set<ExtensionDefinition> extensions =
definitionMap.get(definition);
return (extensions == null) ? Collections.emptySet() : extensions;
}
@Override
- public ConfigurableComponent getTempComponent(final String classType,
final BundleCoordinate bundleCoordinate) {
+ public synchronized ConfigurableComponent getTempComponent(final String
classType, final BundleCoordinate bundleCoordinate) {
if (classType == null) {
throw new IllegalArgumentException("Class type cannot be null");
}
@@ -517,7 +528,29 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
throw new IllegalArgumentException("Bundle Coordinate cannot be
null");
}
- return tempComponentLookup.get(getClassBundleKey(classType,
bundleCoordinate));
+ final String bundleKey = getClassBundleKey(classType,
bundleCoordinate);
+ final ConfigurableComponent existing =
tempComponentLookup.get(bundleKey);
+ if (existing != null) {
+ return existing;
+ }
+
+ final Bundle bundle = getBundle(bundleCoordinate);
+ if (bundle == null) {
+ logger.error("Could not instantiate class of type {} using
ClassLoader for bundle {} because the bundle could not be found", classType,
bundleCoordinate);
+ return null;
+ }
+
+ try {
+ final ClassLoader bundleClassLoader = bundle.getClassLoader();
+ final Class<?> componentClass = Class.forName(classType, true,
bundleClassLoader);
+ final ConfigurableComponent tempComponent =
(ConfigurableComponent) componentClass.newInstance();
+ initializeTempComponent(tempComponent);
+ tempComponentLookup.put(bundleKey, tempComponent);
+ return tempComponent;
+ } catch (final Exception e) {
+ logger.error("Could not instantiate class of type {} using
ClassLoader for bundle {}", classType, bundleCoordinate, e);
+ return null;
+ }
}
private static String getClassBundleKey(final String classType, final
BundleCoordinate bundleCoordinate) {
@@ -529,13 +562,14 @@ public class StandardExtensionDiscoveringManager
implements ExtensionDiscovering
final StringBuilder builder = new StringBuilder();
builder.append("Extension Type Mapping to Bundle:");
- for (final Map.Entry<Class, Set<Class>> entry :
definitionMap.entrySet()) {
+ for (final Map.Entry<Class, Set<ExtensionDefinition>> entry :
definitionMap.entrySet()) {
builder.append("\n\t===
").append(entry.getKey().getSimpleName()).append(" Type ===");
- for (final Class type : entry.getValue()) {
- final List<Bundle> bundles =
classNameBundleLookup.getOrDefault(type.getName(), Collections.emptyList());
+ for (final ExtensionDefinition extensionDefinition :
entry.getValue()) {
+ final String implementationClassName =
extensionDefinition.getImplementationClassName();
+ final List<Bundle> bundles =
classNameBundleLookup.getOrDefault(implementationClassName,
Collections.emptyList());
- builder.append("\n\t").append(type.getName());
+ builder.append("\n\t").append(implementationClassName);
for (final Bundle bundle : bundles) {
final String coordinate =
bundle.getBundleDetails().getCoordinate().getCoordinate();
@@ -599,4 +633,5 @@ public class StandardExtensionDiscoveringManager implements
ExtensionDiscovering
}
}
}
+
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarUnpacker.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarUnpacker.java
index cf41a3a..c5ce5a4 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarUnpacker.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarUnpacker.java
@@ -76,11 +76,11 @@ public final class NarUnpacker {
}
public static ExtensionMapping unpackNars(final Bundle systemBundle, final
File frameworkWorkingDir, final File extensionsWorkingDir, final File
docsWorkingDir, final List<Path> narLibraryDirs) {
- return unpackNars(systemBundle, frameworkWorkingDir,
extensionsWorkingDir, docsWorkingDir, narLibraryDirs, true, true, (coordinate)
-> true);
+ return unpackNars(systemBundle, frameworkWorkingDir,
extensionsWorkingDir, docsWorkingDir, narLibraryDirs, true, true, true,
(coordinate) -> true);
}
public static ExtensionMapping unpackNars(final Bundle systemBundle, final
File frameworkWorkingDir, final File extensionsWorkingDir, final File
docsWorkingDir, final List<Path> narLibraryDirs,
- final boolean
requireFrameworkNar, final boolean requireJettyNar, final
Predicate<BundleCoordinate> narFilter) {
+ final boolean
requireFrameworkNar, final boolean requireJettyNar, final boolean verifyHash,
final Predicate<BundleCoordinate> narFilter) {
final Map<File, BundleCoordinate> unpackedNars = new HashMap<>();
try {
@@ -95,7 +95,10 @@ public final class NarUnpacker {
}
FileUtils.ensureDirectoryExistAndCanReadAndWrite(extensionsWorkingDir);
- FileUtils.ensureDirectoryExistAndCanReadAndWrite(docsWorkingDir);
+
+ if (docsWorkingDir != null) {
+
FileUtils.ensureDirectoryExistAndCanReadAndWrite(docsWorkingDir);
+ }
for (Path narLibraryDir : narLibraryDirs) {
@@ -133,18 +136,18 @@ public final class NarUnpacker {
}
// unpack the framework nar
- unpackedFramework = unpackNar(narFile,
frameworkWorkingDir);
+ unpackedFramework = unpackNar(narFile,
frameworkWorkingDir, verifyHash);
} else if
(NarClassLoaders.JETTY_NAR_ID.equals(bundleCoordinate.getId())) {
if (unpackedJetty != null) {
throw new IllegalStateException("Multiple
Jetty NARs discovered. Only one Jetty NAR is permitted.");
}
// unpack and record the Jetty nar
- unpackedJetty = unpackNar(narFile,
extensionsWorkingDir);
+ unpackedJetty = unpackNar(narFile,
extensionsWorkingDir, verifyHash);
unpackedExtensions.add(unpackedJetty);
} else {
// unpack and record the extension nar
- final File unpackedExtension = unpackNar(narFile,
extensionsWorkingDir);
+ final File unpackedExtension = unpackNar(narFile,
extensionsWorkingDir, verifyHash);
unpackedExtensions.add(unpackedExtension);
}
}
@@ -196,15 +199,6 @@ public final class NarUnpacker {
+ "(" + (int) TimeUnit.SECONDS.convert(duration,
TimeUnit.NANOSECONDS) + " seconds).");
}
- // attempt to delete any docs files that exist so that any
components that have been removed
- // will no longer have entries in the docs folder
- final File[] docsFiles = docsWorkingDir.listFiles();
- if (docsFiles != null) {
- for (final File file : docsFiles) {
- FileUtils.deleteFile(file, true);
- }
- }
-
unpackedNars.putAll(createUnpackedNarBundleCoordinateMap(extensionsWorkingDir));
final ExtensionMapping extensionMapping = new ExtensionMapping();
mapExtensions(unpackedNars, docsWorkingDir, extensionMapping);
@@ -267,9 +261,7 @@ public final class NarUnpacker {
public static void mapExtension(final File unpackedNar, final
BundleCoordinate bundleCoordinate, final File docsDirectory, final
ExtensionMapping mapping) throws IOException {
final File bundledDependencies = new File(unpackedNar,
BUNDLED_DEPENDENCIES_DIRECTORY);
// If docsDirectory is null, assume NiFi is "headless" (no UI or REST
API) and thus no docs are to be generated
- if (docsDirectory != null) {
- unpackBundleDocs(docsDirectory, mapping, bundleCoordinate,
bundledDependencies);
- }
+ unpackBundleDocs(docsDirectory, mapping, bundleCoordinate,
bundledDependencies);
}
private static void unpackBundleDocs(final File docsDirectory, final
ExtensionMapping mapping, final BundleCoordinate bundleCoordinate, final File
bundledDirectory) throws IOException {
@@ -288,16 +280,19 @@ public final class NarUnpacker {
*
* @param nar the nar to unpack
* @param baseWorkingDirectory the directory to unpack to
+ * @param verifyHash if the NAR has already been unpacked, indicates
whether or not the hash should be verified. If this value is true,
+ * and the NAR's hash does not match the hash written to the unpacked
directory, the working directory will be deleted and the NAR will be
+ * unpacked again. If false, the NAR will not be unpacked again and its
hash will not be checked.
* @return the directory to the unpacked NAR
* @throws IOException if unable to explode nar
*/
- public static File unpackNar(final File nar, final File
baseWorkingDirectory) throws IOException {
+ public static File unpackNar(final File nar, final File
baseWorkingDirectory, final boolean verifyHash) throws IOException {
final File narWorkingDirectory = new File(baseWorkingDirectory,
nar.getName() + "-unpacked");
// if the working directory doesn't exist, unpack the nar
if (!narWorkingDirectory.exists()) {
unpack(nar, narWorkingDirectory, FileDigestUtils.getDigest(nar));
- } else {
+ } else if (verifyHash) {
// the working directory does exist. Run digest against the nar
// file and check if the nar has changed since it was deployed.
final byte[] narDigest = FileDigestUtils.getDigest(nar);
@@ -313,6 +308,8 @@ public final class NarUnpacker {
unpack(nar, narWorkingDirectory, narDigest);
}
}
+ } else {
+ logger.debug("Directory {} already exists. Will not verify hash.
Assuming nothing has changed.", narWorkingDirectory);
}
return narWorkingDirectory;
@@ -361,6 +358,10 @@ public final class NarUnpacker {
// merge the extension mapping found in this jar
extensionMapping.merge(jarExtensionMapping);
+ if (docsDirectory == null) {
+ return;
+ }
+
// look for all documentation related to each component
try (final JarFile jarFile = new JarFile(jar)) {
for (final String componentName :
jarExtensionMapping.getAllExtensionNames().keySet()) {
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/StatelessNiFi.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/StatelessNiFi.java
index 925d829..4ae30dc 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/StatelessNiFi.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/StatelessNiFi.java
@@ -69,7 +69,7 @@ public class StatelessNiFi {
logger.info("Unpacking {} NARs", narFiles.length);
final long startUnpack = System.nanoTime();
for (final File narFile : narFiles) {
- NarUnpacker.unpackNar(narFile, narWorkingDirectory);
+ NarUnpacker.unpackNar(narFile, narWorkingDirectory, false);
}
final long millis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startUnpack);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 792e06c..6a386c8 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -123,6 +123,7 @@ import org.apache.nifi.groups.ProcessGroupCounts;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroupCounts;
import org.apache.nifi.history.History;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarClassLoadersHolder;
import org.apache.nifi.parameter.Parameter;
@@ -3100,16 +3101,17 @@ public final class DtoFactory {
/**
* Gets the DocumentedTypeDTOs from the specified classes.
*
- * @param classes classes
+ * @param extensionDefinitions extensionDefinitions
* @param bundleGroupFilter if specified, must be member of bundle group
* @param bundleArtifactFilter if specified, must be member of bundle
artifact
* @param typeFilter if specified, type must match
* @return dtos
*/
- public Set<DocumentedTypeDTO> fromDocumentedTypes(final Set<Class>
classes, final String bundleGroupFilter, final String bundleArtifactFilter,
final String typeFilter) {
+ public Set<DocumentedTypeDTO> fromDocumentedTypes(final
Set<ExtensionDefinition> extensionDefinitions, final String bundleGroupFilter,
final String bundleArtifactFilter, final String typeFilter) {
final Map<Class, Bundle> classBundles = new HashMap<>();
- for (final Class cls : classes) {
- classBundles.put(cls,
extensionManager.getBundle(cls.getClassLoader()));
+ for (final ExtensionDefinition extensionDefinition :
extensionDefinitions) {
+ final Class cls = extensionManager.getClass(extensionDefinition);
+ classBundles.put(cls, extensionDefinition.getBundle());
}
return fromDocumentedTypes(classBundles, bundleGroupFilter,
bundleArtifactFilter, typeFilter);
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
index af10001..e4683fa 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java
@@ -66,6 +66,7 @@ import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.ProcessGroupCounts;
import org.apache.nifi.groups.RemoteProcessGroup;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
@@ -101,8 +102,8 @@ import
org.apache.nifi.web.api.dto.provenance.ProvenanceEventDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceOptionsDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceRequestDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceResultsDTO;
-import org.apache.nifi.web.api.dto.provenance.ProvenanceSearchableFieldDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceSearchValueDTO;
+import org.apache.nifi.web.api.dto.provenance.ProvenanceSearchableFieldDTO;
import org.apache.nifi.web.api.dto.provenance.lineage.LineageDTO;
import org.apache.nifi.web.api.dto.provenance.lineage.LineageRequestDTO;
import
org.apache.nifi.web.api.dto.provenance.lineage.LineageRequestDTO.LineageRequestType;
@@ -515,7 +516,7 @@ public class ControllerFacade implements Authorizable {
public Set<DocumentedTypeDTO> getControllerServiceTypes(final String
serviceType, final String serviceBundleGroup, final String
serviceBundleArtifact, final String serviceBundleVersion,
final String
bundleGroupFilter, final String bundleArtifactFilter, final String typeFilter) {
- final Set<Class> serviceImplementations =
getExtensionManager().getExtensions(ControllerService.class);
+ final Set<ExtensionDefinition> extensionDefinitions =
getExtensionManager().getExtensions(ControllerService.class);
// identify the controller services that implement the specified
serviceType if applicable
if (serviceType != null) {
@@ -538,7 +539,8 @@ public class ControllerFacade implements Authorizable {
final Map<Class, Bundle> matchingServiceImplementations = new
HashMap<>();
// check each type and remove those that aren't in the specified
ancestry
- for (final Class csClass : serviceImplementations) {
+ for (final ExtensionDefinition extensionDefinition :
extensionDefinitions) {
+ final Class csClass =
getExtensionManager().getClass(extensionDefinition);
if (implementsServiceType(serviceClass, csClass)) {
matchingServiceImplementations.put(csClass,
getExtensionManager().getBundle(csClass.getClassLoader()));
}
@@ -546,7 +548,7 @@ public class ControllerFacade implements Authorizable {
return
dtoFactory.fromDocumentedTypes(matchingServiceImplementations,
bundleGroupFilter, bundleArtifactFilter, typeFilter);
} else {
- return dtoFactory.fromDocumentedTypes(serviceImplementations,
bundleGroupFilter, bundleArtifactFilter, typeFilter);
+ return dtoFactory.fromDocumentedTypes(extensionDefinitions,
bundleGroupFilter, bundleArtifactFilter, typeFilter);
}
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-bootstrap/src/main/java/org/apache/nifi/stateless/bootstrap/StatelessBootstrap.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-bootstrap/src/main/java/org/apache/nifi/stateless/bootstrap/StatelessBootstrap.java
index 8357115..415e1db 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-bootstrap/src/main/java/org/apache/nifi/stateless/bootstrap/StatelessBootstrap.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-bootstrap/src/main/java/org/apache/nifi/stateless/bootstrap/StatelessBootstrap.java
@@ -47,6 +47,9 @@ import java.util.regex.Pattern;
public class StatelessBootstrap {
private static final Logger logger =
LoggerFactory.getLogger(StatelessBootstrap.class);
private static final Pattern STATELESS_NAR_PATTERN =
Pattern.compile("nifi-stateless-nar-.*\\.nar-unpacked");
+ private static final String NIFI_GROUP = "org.apache.nifi";
+ private static final String NIFI_STATELESS_ARTIFACT_ID =
"nifi-stateless-nar";
+ private static final String NIFI_JETTY_ARTIFACT_ID = "nifi-jetty-bundle";
private final ClassLoader statelessClassLoader;
private final StatelessEngineConfiguration engineConfiguration;
@@ -87,17 +90,15 @@ public class StatelessBootstrap {
throw new IOException("Working Directory " + workingDirectory + "
does not exist and could not be created");
}
-
final Bundle systemBundle =
SystemBundle.create(narDirectory.getAbsolutePath(),
ClassLoader.getSystemClassLoader());
final File frameworkWorkingDir = new File(workingDirectory,
"nifi-framework");
final File extensionsWorkingDir = new File(workingDirectory,
"extensions");
- final File docsWorkingDir = new File(workingDirectory,
"documentation");
final List<Path> narDirectories =
Collections.singletonList(narDirectory.toPath());
// Unpack NARs
final long unpackStart = System.currentTimeMillis();
final Predicate<BundleCoordinate> narFilter = coordinate -> true;
- NarUnpacker.unpackNars(systemBundle, frameworkWorkingDir,
extensionsWorkingDir, docsWorkingDir, narDirectories, false, false, narFilter);
+ NarUnpacker.unpackNars(systemBundle, frameworkWorkingDir,
extensionsWorkingDir, null, narDirectories, false, false, false, narFilter);
final long unpackMillis = System.currentTimeMillis() - unpackStart;
logger.info("Unpacked NAR files in {} millis", unpackMillis);
@@ -121,6 +122,16 @@ public class StatelessBootstrap {
return new StatelessBootstrap(statelessClassLoader,
engineConfiguration);
}
+ private static boolean isRequiredForBootstrap(final BundleCoordinate
coordinate) {
+ final String group = coordinate.getGroup();
+ if (!NIFI_GROUP.equals(group)) {
+ return false;
+ }
+
+ final String artifactId = coordinate.getId();
+ return NIFI_JETTY_ARTIFACT_ID.equals(artifactId) ||
NIFI_STATELESS_ARTIFACT_ID.equals(artifactId);
+ }
+
private static File locateStatelessNarWorkingDirectory(final File
workingDirectory) throws IOException {
final File[] files = workingDirectory.listFiles();
if (files == null) {
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/extensions/FileSystemExtensionRepository.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/extensions/FileSystemExtensionRepository.java
index 7aa1ca5..1f9ddbf 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/extensions/FileSystemExtensionRepository.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/extensions/FileSystemExtensionRepository.java
@@ -51,8 +51,8 @@ public class FileSystemExtensionRepository implements
ExtensionRepository {
private final List<ExtensionClient> clients;
- public FileSystemExtensionRepository(final ExtensionDiscoveringManager
extensionManager, final File narLibDirectory, final File workingDirectory,
final NarClassLoaders narClassLoaders,
- final List<ExtensionClient> clients) {
+ public FileSystemExtensionRepository(final ExtensionDiscoveringManager
extensionManager, final File narLibDirectory, final File workingDirectory,
+ final NarClassLoaders
narClassLoaders, final List<ExtensionClient> clients) {
this.extensionManager = extensionManager;
this.narLibDirectory = narLibDirectory;
this.workingDirectory = workingDirectory;
@@ -118,7 +118,7 @@ public class FileSystemExtensionRepository implements
ExtensionRepository {
// even if they use a different ExtensionRepository.
unpackLock.lock();
try {
- final File unpackedDir = NarUnpacker.unpackNar(downloadedFile,
workingDirectory);
+ final File unpackedDir = NarUnpacker.unpackNar(downloadedFile,
workingDirectory, false);
unpackedDirs.add(unpackedDir);
} finally {
unpackLock.unlock();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/bootstrap/ExtensionDiscovery.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/bootstrap/ExtensionDiscovery.java
index cb7b7fb..9233b89 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/bootstrap/ExtensionDiscovery.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/bootstrap/ExtensionDiscovery.java
@@ -32,7 +32,8 @@ public class ExtensionDiscovery {
private static final Logger logger =
LoggerFactory.getLogger(ExtensionDiscovery.class);
public static ExtensionDiscoveringManager discover(final File
narWorkingDirectory, final ClassLoader systemClassLoader, final NarClassLoaders
narClassLoaders) throws IOException {
- final long discoveryStart = System.nanoTime();
+ logger.info("Initializing NAR ClassLoaders");
+
try {
narClassLoaders.init(systemClassLoader, null, narWorkingDirectory);
} catch (final ClassNotFoundException cnfe) {
@@ -41,9 +42,9 @@ public class ExtensionDiscovery {
final Set<Bundle> narBundles = narClassLoaders.getBundles();
+ final long discoveryStart = System.nanoTime();
final StandardExtensionDiscoveringManager extensionManager = new
StandardExtensionDiscoveringManager();
extensionManager.discoverExtensions(narBundles);
- extensionManager.logClassLoaderMapping();
final long discoveryMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - discoveryStart);
logger.info("Successfully discovered extensions in {} milliseconds",
discoveryMillis);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StandardStatelessEngine.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StandardStatelessEngine.java
index 7094dbe..19fc7b9 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StandardStatelessEngine.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StandardStatelessEngine.java
@@ -17,6 +17,7 @@
package org.apache.nifi.stateless.engine;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.AllowableValue;
@@ -36,6 +37,7 @@ import org.apache.nifi.encrypt.PropertyEncryptor;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.extensions.ExtensionRepository;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.parameter.Parameter;
import org.apache.nifi.parameter.ParameterContext;
@@ -83,7 +85,7 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
private final ExtensionManager extensionManager;
private final BulletinRepository bulletinRepository;
private final StatelessStateManagerProvider stateManagerProvider;
- private final PropertyEncryptor encryptor;
+ private final PropertyEncryptor propertyEncryptor;
private final FlowRegistryClient flowRegistryClient;
private final VariableRegistry rootVariableRegistry;
private final ProcessScheduler processScheduler;
@@ -108,7 +110,7 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
this.extensionManager = requireNonNull(builder.extensionManager,
"Extension Manager must be provided");
this.bulletinRepository = requireNonNull(builder.bulletinRepository,
"Bulletin Repository must be provided");
this.stateManagerProvider =
requireNonNull(builder.stateManagerProvider, "State Manager Provider must be
provided");
- this.encryptor = requireNonNull(builder.encryptor, "Encryptor must be
provided");
+ this.propertyEncryptor = requireNonNull(builder.propertyEncryptor,
"Encryptor must be provided");
this.flowRegistryClient = requireNonNull(builder.flowRegistryClient,
"Flow Registry Client must be provided");
this.rootVariableRegistry = requireNonNull(builder.variableRegistry,
"Variable Registry must be provided");
this.processScheduler = requireNonNull(builder.processScheduler,
"Process Scheduler must be provided");
@@ -343,12 +345,14 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
final Set<String> possibleResolvedClassNames = new HashSet<>();
- final Set<Class> implementationClasses =
extensionManager.getExtensions(ReportingTask.class);
- for (final Class<?> implementationClass : implementationClasses) {
- if (implementationClass.getSimpleName().equals(specifiedType)) {
- logger.debug("Found possible matching class {}",
implementationClass);
+ final Set<ExtensionDefinition> definitions =
extensionManager.getExtensions(ReportingTask.class);
+ for (final ExtensionDefinition definition : definitions) {
+ final String implementationClassName =
definition.getImplementationClassName();
+ final String simpleName = implementationClassName.contains(".") ?
StringUtils.substringAfterLast(implementationClassName, ".") :
implementationClassName;
+ if (simpleName.equals(specifiedType)) {
+ logger.debug("Found possible matching class {}",
implementationClassName);
- possibleResolvedClassNames.add(implementationClass.getName());
+ possibleResolvedClassNames.add(implementationClassName);
}
}
@@ -438,46 +442,57 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
logger.info("Registered Parameter Context {}",
parameterContextDefinition.getName());
}
+ @Override
public ExtensionManager getExtensionManager() {
return extensionManager;
}
+ @Override
public BulletinRepository getBulletinRepository() {
return bulletinRepository;
}
+ @Override
public StateManagerProvider getStateManagerProvider() {
return stateManagerProvider;
}
- public PropertyEncryptor getEncryptor() {
- return encryptor;
+ @Override
+ public PropertyEncryptor getPropertyEncryptor() {
+ return propertyEncryptor;
}
+ @Override
public FlowRegistryClient getFlowRegistryClient() {
return flowRegistryClient;
}
+ @Override
public VariableRegistry getRootVariableRegistry() {
return rootVariableRegistry;
}
+ @Override
public ProcessScheduler getProcessScheduler() {
return processScheduler;
}
+ @Override
public ReloadComponent getReloadComponent() {
return reloadComponent;
}
+ @Override
public ControllerServiceProvider getControllerServiceProvider() {
return controllerServiceProvider;
}
+ @Override
public ProvenanceRepository getProvenanceRepository() {
return provenanceRepository;
}
+ @Override
public FlowFileEventRepository getFlowFileEventRepository() {
return flowFileEventRepository;
}
@@ -497,7 +512,7 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
private ExtensionManager extensionManager = null;
private BulletinRepository bulletinRepository = null;
private StatelessStateManagerProvider stateManagerProvider = null;
- private PropertyEncryptor encryptor = null;
+ private PropertyEncryptor propertyEncryptor = null;
private FlowRegistryClient flowRegistryClient = null;
private VariableRegistry variableRegistry = null;
private ProcessScheduler processScheduler = null;
@@ -521,8 +536,8 @@ public class StandardStatelessEngine implements
StatelessEngine<VersionedFlowSna
return this;
}
- public Builder encryptor(final PropertyEncryptor encryptor) {
- this.encryptor = encryptor;
+ public Builder encryptor(final PropertyEncryptor propertyEncryptor) {
+ this.propertyEncryptor = propertyEncryptor;
return this;
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessEngine.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessEngine.java
index 00d97c2..31213a5 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessEngine.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessEngine.java
@@ -48,7 +48,7 @@ public interface StatelessEngine<T> {
StateManagerProvider getStateManagerProvider();
- PropertyEncryptor getEncryptor();
+ PropertyEncryptor getPropertyEncryptor();
FlowRegistryClient getFlowRegistryClient();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
index 89bf971..95e9437 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
@@ -209,7 +209,7 @@ public class StatelessFlowManager extends
AbstractFlowManager implements FlowMan
return new StandardProcessGroup(id,
statelessEngine.getControllerServiceProvider(),
statelessEngine.getProcessScheduler(),
- statelessEngine.getEncryptor(),
+ statelessEngine.getPropertyEncryptor(),
statelessEngine.getExtensionManager(),
statelessEngine.getStateManagerProvider(),
this,
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessDataflowFactory.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessDataflowFactory.java
index 6621c9d..5832896 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessDataflowFactory.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessDataflowFactory.java
@@ -148,7 +148,28 @@ public class StandardStatelessDataflowFactory implements
StatelessDataflowFactor
narClassLoaders, extensionClients);
final VariableRegistry variableRegistry =
VariableRegistry.EMPTY_REGISTRY;
- final PropertyEncryptor encryptor =
getPropertyEncryptor(engineConfiguration.getSensitivePropsKey());
+ final PropertyEncryptor lazyInitializedEncryptor = new
PropertyEncryptor() {
+ private PropertyEncryptor created = null;
+
+ @Override
+ public String encrypt(final String property) {
+ return getEncryptor().encrypt(property);
+ }
+
+ @Override
+ public String decrypt(final String encryptedProperty) {
+ return getEncryptor().decrypt(encryptedProperty);
+ }
+
+ private synchronized PropertyEncryptor getEncryptor() {
+ if (created != null) {
+ return created;
+ }
+
+ created =
getPropertyEncryptor(engineConfiguration.getSensitivePropsKey());
+ return created;
+ }
+ };
final File krb5File = engineConfiguration.getKrb5File();
final KerberosConfig kerberosConfig = new KerberosConfig(null,
null, krb5File);
@@ -157,7 +178,7 @@ public class StandardStatelessDataflowFactory implements
StatelessDataflowFactor
final StatelessEngine<VersionedFlowSnapshot> statelessEngine = new
StandardStatelessEngine.Builder()
.bulletinRepository(bulletinRepository)
- .encryptor(encryptor)
+ .encryptor(lazyInitializedEncryptor)
.extensionManager(extensionManager)
.flowRegistryClient(flowRegistryClient)
.stateManagerProvider(stateManagerProvider)
@@ -172,7 +193,7 @@ public class StandardStatelessDataflowFactory implements
StatelessDataflowFactor
final StatelessFlowManager flowManager = new
StatelessFlowManager(flowFileEventRepo, parameterContextManager,
statelessEngine, () -> true, sslContext);
final ControllerServiceProvider controllerServiceProvider = new
StandardControllerServiceProvider(processScheduler, bulletinRepository,
flowManager, extensionManager);
- final ProcessContextFactory rawProcessContextFactory = new
StatelessProcessContextFactory(controllerServiceProvider, encryptor,
stateManagerProvider);
+ final ProcessContextFactory rawProcessContextFactory = new
StatelessProcessContextFactory(controllerServiceProvider,
lazyInitializedEncryptor, stateManagerProvider);
final ProcessContextFactory processContextFactory = new
CachingProcessContextFactory(rawProcessContextFactory);
contentRepo = new ByteArrayContentRepository();
flowFileRepo = new StatelessFlowFileRepository();
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlow.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlow.java
index 4edea6e..672a79b 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlow.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/flow/StandardStatelessFlow.java
@@ -39,7 +39,6 @@ import
org.apache.nifi.controller.repository.StandardProcessSessionFactory;
import org.apache.nifi.controller.repository.metrics.StandardFlowFileEvent;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
-import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.controller.state.StandardStateMap;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.groups.ProcessGroup;
@@ -47,6 +46,7 @@ import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.exception.TerminatedTaskException;
import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.stateless.engine.ExecutionProgress;
@@ -120,6 +120,7 @@ public class StandardStatelessFlow implements
StatelessDataflow {
internalFlowFileQueues = discoverInternalFlowFileQueues(rootGroup);
}
+
private List<FlowFileQueue> discoverInternalFlowFileQueues(final
ProcessGroup group) {
final Set<Port> rootGroupInputPorts = rootGroup.getInputPorts();
final Set<Port> rootGroupOutputPorts = rootGroup.getOutputPorts();
@@ -223,38 +224,28 @@ public class StandardStatelessFlow implements
StatelessDataflow {
final long startTime = System.currentTimeMillis();
final long cutoff = startTime + COMPONENT_ENABLE_TIMEOUT_MILLIS;
- while (isAnyServiceEnabling(group)) {
- if (System.currentTimeMillis() > cutoff) {
- final String validationErrors = performValidation().toString();
- throw new IllegalStateException("At least one Controller
Service never finished enabling. All validation errors: " + validationErrors);
- }
-
- logger.debug("At least one Controller Service in group {} is still
enabling. Will wait 5 milliseconds and check again", group);
-
+ final Set<ControllerServiceNode> serviceNodes =
group.findAllControllerServices();
+ for (final ControllerServiceNode serviceNode : serviceNodes) {
+ final boolean enabled;
try {
- Thread.sleep(5L);
+ enabled = serviceNode.awaitEnabled(cutoff -
System.currentTimeMillis(), TimeUnit.MILLISECONDS);
} catch (final InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while waiting for
Controller Services to enable", ie);
}
- }
- for (final ProcessGroup childGroup : group.getProcessGroups()) {
- waitForServicesEnabled(childGroup);
- }
- }
+ if (enabled) {
+ continue;
+ }
- private boolean isAnyServiceEnabling(final ProcessGroup group) {
- for (final ControllerServiceNode serviceNode :
group.getControllerServices(false)) {
- final ControllerServiceState state = serviceNode.getState();
- if (state == ControllerServiceState.ENABLING) {
- return true;
+ if (System.currentTimeMillis() > cutoff) {
+ final String validationErrors = performValidation().toString();
+ throw new IllegalStateException("At least one Controller
Service never finished enabling. All validation errors: " + validationErrors);
}
}
-
- return false;
}
+
private void startReportingTasks() {
reportingTasks.forEach(this::startReportingTask);
}
@@ -433,15 +424,13 @@ public class StandardStatelessFlow implements
StatelessDataflow {
final long start = System.nanoTime();
final long processingNanos;
- int invocations = 0;
// If there is no incoming connection, trigger once.
logger.debug("Triggering {}", connectable);
connectable.onTrigger(processContext, sessionFactory);
- invocations = 1;
processingNanos = System.nanoTime() - start;
- registerProcessEvent(connectable, invocations,
processingNanos);
+ registerProcessEvent(connectable, 1, processingNanos);
}
} catch (final TerminatedTaskException tte) {
// This occurs when the caller invokes the cancel() method of
DataflowTrigger.
@@ -624,6 +613,13 @@ public class StandardStatelessFlow implements
StatelessDataflow {
return latest;
}
+ @SuppressWarnings("unused")
+ public Set<Processor> findAllProcessors() {
+ return rootGroup.findAllProcessors().stream()
+ .map(ProcessorNode::getProcessor)
+ .collect(Collectors.toSet());
+ }
+
private static class SerializableStateMap {
private long version;
private Map<String, String> stateValues;
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenTCP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenTCP.java
index 83f4a75..4091516 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenTCP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenTCP.java
@@ -16,15 +16,6 @@
*/
package org.apache.nifi.processors.standard;
-import java.io.IOException;
-import java.net.Socket;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import javax.net.SocketFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLException;
-
import org.apache.commons.io.IOUtils;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSessionFactory;
@@ -43,6 +34,14 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
public class TestListenTCP {
private static final long RESPONSE_TIMEOUT = 10000;
@@ -156,7 +155,7 @@ public class TestListenTCP {
messages.add("This is message 5\n");
// Make an SSLContext that only has the trust store, this should not
work since the processor has client auth REQUIRED
- Assert.assertThrows(SSLException.class, () ->
+ Assert.assertThrows(IOException.class, () ->
runTCP(messages, messages.size(), trustStoreSslContext)
);
}
diff --git
a/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/StatelessSystemIT.java
b/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/StatelessSystemIT.java
index db200cf..4fe6c76 100644
---
a/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/StatelessSystemIT.java
+++
b/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/StatelessSystemIT.java
@@ -50,7 +50,7 @@ public class StatelessSystemIT {
// We reference version 1.13.0 here, but the version isn't really
relevant. Because there will only be a single artifact of name
"nifi-system-test-extensions-nar" the framework will end
// up finding a "compatible bundle" and using that, regardless of the
specified version.
- protected static final Bundle SYSTEM_TEST_EXTENSIONS_BUNDLE = new
Bundle("org.apache.nifi", "nifi-system-test-extensions-nar", "1.13.0");
+ protected static final Bundle SYSTEM_TEST_EXTENSIONS_BUNDLE = new
Bundle("org.apache.nifi", "nifi-system-test-extensions-nar", "1.13.0-SNAPSHOT");
@Rule
public TestName name = new TestName();
diff --git
a/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/classloader/InstanceClassLoaderIT.java
b/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/classloader/InstanceClassLoaderIT.java
new file mode 100644
index 0000000..6268598
--- /dev/null
+++
b/nifi-system-tests/nifi-stateless-system-test-suite/src/test/java/org/apache/nifi/stateless/classloader/InstanceClassLoaderIT.java
@@ -0,0 +1,100 @@
+/*
+ * 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.stateless.classloader;
+
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.registry.flow.VersionedProcessor;
+import org.apache.nifi.stateless.StatelessSystemIT;
+import org.apache.nifi.stateless.VersionedFlowBuilder;
+import org.apache.nifi.stateless.config.StatelessConfigurationException;
+import org.apache.nifi.stateless.flow.StatelessDataflow;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+
+public class InstanceClassLoaderIT extends StatelessSystemIT {
+
+ @Test
+ public void testProcessorHasNarClassLoader() throws IOException,
StatelessConfigurationException, NoSuchMethodException, IllegalAccessException,
InvocationTargetException {
+ final VersionedFlowBuilder builder = new VersionedFlowBuilder();
+ final VersionedProcessor passThrough1 =
builder.createSimpleProcessor("PassThrough");
+
passThrough1.setAutoTerminatedRelationships(Collections.singleton("success"));
+
+ final VersionedProcessor passThrough2 =
builder.createSimpleProcessor("PassThrough");
+
passThrough2.setAutoTerminatedRelationships(Collections.singleton("success"));
+
+ // Create the flow
+ final StatelessDataflow dataflow =
loadDataflow(builder.getFlowSnapshot());
+ final Set<Processor> processors = findProcessors(dataflow);
+ assertEquals(2, processors.size());
+
+ final Set<ClassLoader> classLoaders = new HashSet<>();
+ for (final Processor processor : processors) {
+ final ClassLoader classLoader =
processor.getClass().getClassLoader();
+ classLoaders.add(classLoader);
+ assertEquals("org.apache.nifi.nar.NarClassLoader",
classLoader.getClass().getName());
+ }
+
+ assertEquals(1, classLoaders.size());
+ }
+
+ @Test
+ public void testProcessorHasInstanceClassLoader() throws IOException,
StatelessConfigurationException, NoSuchMethodException, IllegalAccessException,
InvocationTargetException {
+ final VersionedFlowBuilder builder = new VersionedFlowBuilder();
+ final VersionedProcessor passThrough1 =
builder.createSimpleProcessor("PassThroughRequiresInstanceClassLoading");
+
passThrough1.setAutoTerminatedRelationships(Collections.singleton("success"));
+
+ final VersionedProcessor passThrough2 =
builder.createSimpleProcessor("PassThroughRequiresInstanceClassLoading");
+
passThrough2.setAutoTerminatedRelationships(Collections.singleton("success"));
+
+ // Create the flow
+ final StatelessDataflow dataflow =
loadDataflow(builder.getFlowSnapshot());
+ final Set<Processor> processors = findProcessors(dataflow);
+ assertEquals(2, processors.size());
+
+ final Set<ClassLoader> classLoaders = new HashSet<>();
+ for (final Processor processor : processors) {
+ final ClassLoader classLoader =
processor.getClass().getClassLoader();
+ classLoaders.add(classLoader);
+ assertEquals("org.apache.nifi.nar.InstanceClassLoader",
classLoader.getClass().getName());
+ }
+
+ assertEquals(2, classLoaders.size());
+
+ final Set<ClassLoader> parentClassLoaders = classLoaders.stream()
+ .map(ClassLoader::getParent)
+ .collect(Collectors.toSet());
+
+ assertEquals(1, parentClassLoaders.size());
+ }
+
+ private Set<Processor> findProcessors(final StatelessDataflow dataflow)
throws NoSuchMethodException, InvocationTargetException, IllegalAccessException
{
+ final Method method =
dataflow.getClass().getDeclaredMethod("findAllProcessors");
+ method.setAccessible(true);
+ return (Set<Processor>) method.invoke(dataflow);
+ }
+
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThrough.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThrough.java
new file mode 100644
index 0000000..cc4fb4a
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThrough.java
@@ -0,0 +1,51 @@
+/*
+ * 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.processors.tests.system;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import java.util.Collections;
+import java.util.Set;
+
+public class PassThrough extends AbstractProcessor {
+
+ private static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Everything goes here")
+ .build();
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Collections.singleton(REL_SUCCESS);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThroughRequiresInstanceClassLoading.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThroughRequiresInstanceClassLoading.java
new file mode 100644
index 0000000..4c0cd29
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/PassThroughRequiresInstanceClassLoading.java
@@ -0,0 +1,53 @@
+/*
+ * 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.processors.tests.system;
+
+import org.apache.nifi.annotation.behavior.RequiresInstanceClassLoading;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import java.util.Collections;
+import java.util.Set;
+
+@RequiresInstanceClassLoading(cloneAncestorResources = true)
+public class PassThroughRequiresInstanceClassLoading extends AbstractProcessor
{
+
+ private static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Everything goes here")
+ .build();
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Collections.singleton(REL_SUCCESS);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 169a9c6..c6588f6 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -21,6 +21,8 @@
org.apache.nifi.processors.tests.system.EvaluatePropertiesWithDifferentELScopes
org.apache.nifi.processors.tests.system.FakeProcessor
org.apache.nifi.processors.tests.system.FakeDynamicPropertiesProcessor
org.apache.nifi.processors.tests.system.GenerateFlowFile
+org.apache.nifi.processors.tests.system.PassThrough
+org.apache.nifi.processors.tests.system.PassThroughRequiresInstanceClassLoading
org.apache.nifi.processors.tests.system.ReverseContents
org.apache.nifi.processors.tests.system.SetAttribute
org.apache.nifi.processors.tests.system.Sleep
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithDifferentFlow.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithDifferentFlow.java
index c37abe2..386fc35 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithDifferentFlow.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/JoinClusterWithDifferentFlow.java
@@ -55,7 +55,6 @@ import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.ParserConfigurationException;
import java.io.ByteArrayOutputStream;
import java.io.File;
-import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
@@ -107,16 +106,16 @@ public class JoinClusterWithDifferentFlow extends
NiFiSystemIT {
}
- private File getBackupFile(final File confDir) throws InterruptedException
{
- final FileFilter fileFilter = file ->
file.getName().startsWith("flow") && file.getName().endsWith(".xml.gz");
+ private List<File> getFlowXmlFiles(final File confDir) {
+ final File[] flowXmlFileArray = confDir.listFiles(file ->
file.getName().startsWith("flow") && file.getName().endsWith(".xml.gz"));
+ final List<File> flowXmlFiles = new
ArrayList<>(Arrays.asList(flowXmlFileArray));
+ return flowXmlFiles;
+ }
- waitFor(() -> {
- final File[] flowXmlFileArray = confDir.listFiles(fileFilter);
- return flowXmlFileArray != null && flowXmlFileArray.length == 2;
- });
+ private File getBackupFile(final File confDir) throws InterruptedException
{
+ waitFor(() -> getFlowXmlFiles(confDir).size() == 2);
- final File[] flowXmlFileArray = confDir.listFiles(fileFilter);
- final List<File> flowXmlFiles = new
ArrayList<>(Arrays.asList(flowXmlFileArray));
+ final List<File> flowXmlFiles = getFlowXmlFiles(confDir);
assertEquals(2, flowXmlFiles.size());
flowXmlFiles.removeIf(file -> file.getName().equals("flow.xml.gz"));
