Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-10383: When importing/synching to a VersionedFlow, if the Versioned PropertyDescriptor indicates that a property references a Controller Service, treat it the same as if the PropertyDescriptor itself indicates that it references a service. This allows us to ensure that scripted components' property descriptors that reference controller services are properly mapped. Additionally updated StandardProcessorNode so that when we have this condition that we properly account for the Controller Service reference 

Signed-off-by: Matthew Burgess <mattyb149@apache.org>

This closes #6322
This commit is contained in:
Mark Payne 2022-08-22 15:13:43 -04:00 committed by Matthew Burgess
parent cb275dd956
commit 9070cd0904
No known key found for this signature in database
GPG Key ID: 05D3DEB8126DAD24
2 changed files with 42 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
View File

@ -39,6 +39,7 @@ import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.ConfigVerificationResult.Outcome;
import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.validation.ValidationState;
@ -2079,6 +2080,35 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getExtensionManager(), getProcessor().getClass(), getProcessor().getIdentifier())) {
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnConfigurationRestored.class, getProcessor(), context);
}
// Now that the configuration has been fully restored, it's possible that some components (in particular, scripted components)
// may change the Property Descriptor definitions because they've now reloaded the configured scripts. As a result,
// the PropertyDescriptor may now indicate that it references a Controller Service when it previously had not.
// In order to account for this, we need to refresh our controller service references by removing an previously existing
// references and establishing new references.
updateControllerServiceReferences();
}
private void updateControllerServiceReferences() {
for (final Map.Entry<PropertyDescriptor, PropertyConfiguration> entry : getProperties().entrySet()) {
final PropertyDescriptor descriptor = entry.getKey();
final PropertyConfiguration propertyConfiguration = entry.getValue();
if (descriptor.getControllerServiceDefinition() == null || propertyConfiguration == null) {
continue;
}
final String propertyValue = propertyConfiguration.getEffectiveValue(getParameterLookup());
if (propertyValue == null) {
continue;
}
final ControllerServiceNode serviceNode = getControllerServiceProvider().getControllerServiceNode(propertyValue);
if (serviceNode == null) {
continue;
}
serviceNode.removeReference(this, descriptor);
serviceNode.addReference(this, descriptor);
}
}
}

19
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
View File

@ -1206,7 +1206,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
}
final Set<String> sensitiveDynamicPropertyNames = getSensitiveDynamicPropertyNames(service, proposed.getProperties(), proposed.getPropertyDescriptors().values());
final Map<String, String> properties = populatePropertiesMap(service, proposed.getProperties(), service.getProcessGroup());
final Map<String, String> properties = populatePropertiesMap(service, proposed.getProperties(), proposed.getPropertyDescriptors(), service.getProcessGroup());
service.setProperties(properties, true, sensitiveDynamicPropertyNames);
if (!isEqual(service.getBundleCoordinate(), proposed.getBundle())) {
@ -1249,7 +1249,8 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
return sensitiveDynamicPropertyNames;
}
private Map<String, String> populatePropertiesMap(final ComponentNode componentNode, final Map<String, String> proposedProperties, final ProcessGroup group) {
private Map<String, String> populatePropertiesMap(final ComponentNode componentNode, final Map<String, String> proposedProperties,
final Map<String, VersionedPropertyDescriptor> proposedPropertyDescriptors, final ProcessGroup group) {
// Explicitly set all existing properties to null, except for sensitive properties, so that if there isn't an entry in the proposedProperties
// it will get removed from the processor. We don't do this for sensitive properties because when we retrieve the VersionedProcessGroup from registry,
@ -1270,9 +1271,14 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
for (final String propertyName : updatedPropertyNames) {
final PropertyDescriptor descriptor = componentNode.getPropertyDescriptor(propertyName);
final VersionedPropertyDescriptor versionedDescriptor = (proposedPropertyDescriptors == null) ? null : proposedPropertyDescriptors.get(propertyName);
final boolean referencesService = (descriptor != null && descriptor.getControllerServiceDefinition() != null)
|| (versionedDescriptor != null && versionedDescriptor.getIdentifiesControllerService());
final boolean sensitive = (descriptor != null && descriptor.isSensitive())
|| (versionedDescriptor != null && versionedDescriptor.isSensitive());
String value;
if (descriptor != null && descriptor.getControllerServiceDefinition() != null ) {
if (descriptor != null && referencesService) {
// Need to determine if the component's property descriptor for this service is already set to an id
// of an existing service that is outside the current processor group, and if it is we want to leave
// the property set to that value
@ -1293,11 +1299,10 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
if (existingExternalServiceId == null) {
final String serviceVersionedComponentId = proposedProperties.get(propertyName);
String instanceId = getServiceInstanceId(serviceVersionedComponentId, group);
value = instanceId == null ? serviceVersionedComponentId : instanceId;
value = (instanceId == null) ? serviceVersionedComponentId : instanceId;
} else {
value = existingExternalServiceId;
}
} else {
value = proposedProperties.get(propertyName);
}
@ -1307,7 +1312,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
// populated value. The exception to this rule is if the currently configured value is a Parameter Reference and the Versioned Flow is empty. In this case, it implies
// that the Versioned Flow has changed from a Parameter Reference to an explicit value. In this case, we do in fact want to change the value of the Sensitive Property from
// the current parameter reference to an unset value.
if (descriptor.isSensitive() && value == null) {
if (sensitive && value == null) {
final PropertyConfiguration propertyConfiguration = componentNode.getProperty(descriptor);
if (propertyConfiguration == null) {
continue;
@ -2533,7 +2538,7 @@ public class StandardVersionedComponentSynchronizer implements VersionedComponen
processor.setPenalizationPeriod(proposed.getPenaltyDuration());
final Set<String> sensitiveDynamicPropertyNames = getSensitiveDynamicPropertyNames(processor, proposed.getProperties(), proposed.getPropertyDescriptors().values());
final Map<String, String> properties = populatePropertiesMap(processor, proposed.getProperties(), processor.getProcessGroup());
final Map<String, String> properties = populatePropertiesMap(processor, proposed.getProperties(), proposed.getPropertyDescriptors(), processor.getProcessGroup());
processor.setProperties(properties, true, sensitiveDynamicPropertyNames);
processor.setRunDuration(proposed.getRunDurationMillis(), TimeUnit.MILLISECONDS);
processor.setSchedulingStrategy(SchedulingStrategy.valueOf(proposed.getSchedulingStrategy()));